clean files

stop_machines.py

@@ -1,17 +0,0 @@
-import requests, os
-from dotenv import load_dotenv
-
-load_dotenv()
-user_id = os.getenv("RUNCOMFY_USER_ID")
-api_token = os.getenv("API_TOKEN")
-
-s = requests.Session()
-h = {"Authorization": f"Bearer {api_token}"}
-url = f"https://beta-api.runcomfy.net/prod/api/users/{user_id}/servers"
-
-servers = s.get(url, headers=h).json()
-print("Found servers:", len(servers))
-for sv in servers:
-    print("Deleting", sv["server_id"])
-    s.delete(f"{url}/{sv['server_id']}", headers=h)
-print("Done")

test_e2e_proxy.py

@@ -1,124 +0,0 @@
-import os
-import sys
-import json
-import time
-import requests
-import asyncio
-
-# 添加项目根目录到 Python 路径
-sys.path.insert(0, str(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
-from agent.tools.builtin.toolhub import _preprocess_params
-
-ROUTER_URL = "http://127.0.0.1:8001/run_tool"
-
-def call_tool(tool_id, params):
-    print(f"\n[{tool_id}] Calling...")
-    # 就像真正的 agent 流程一样，先预处理参数（将本地文件转化为 CDN 链接）
-    processed_params = asyncio.run(_preprocess_params(params))
-    
-    resp = requests.post(ROUTER_URL, json={"tool_id": tool_id, "params": processed_params})
-    resp.raise_for_status()
-    res = resp.json()
-    if res.get("status") == "error":
-        raise Exception(f"Tool error: {res.get('error')}")
-    print(f"[{tool_id}] Success")
-    return res.get("result", {})
-
-def test_full_workflow():
-    print("=== RunComfy Workflow E2E Test ===")
-    
-    # 0. Check existing machines
-    print("\n0. Checking for existing ComfyUI Instances...")
-    status_res = call_tool("runcomfy_check_status", {})
-    servers = status_res.get("servers", [])
-    
-    server_id = None
-    for s in servers:
-        if s.get("current_status") == "Ready":
-            server_id = s.get("server_id")
-            print(f"Found Ready machine: {server_id}")
-            break
-        elif s.get("current_status") in ("Starting", "Creating") and not server_id:
-            server_id = s.get("server_id")
-            print(f"Found Starting machine: {server_id}. Note: You might need to wait for it to be Ready.")
-
-    if not server_id:
-        raise Exception("\nNo active machine found. Expecting an already running machine. Aborting.")
-    
-    print("\n1. Proceeding with existing machine...")
-
-    try:
-        print("\n2. Loading custom workflow from JSON...")
-        import os
-        workflow_path = os.path.join(os.path.dirname(__file__), "flux_depth_controlnet_workflow.json")
-        with open(workflow_path, "r", encoding="utf-8") as f:
-            workflow_api = json.load(f)
-
-        print("\n3. Executing workflow with ControlNet inputs...")
-        
-        # 收集输入图片文件列表传递给 executor
-        # 我们在这里填入包含本地路径的 url
-        # call_tool 会使用 ToolHub 的 _preprocess_params 自动把这些 url 的本地路径替换为真正的 CDN 链接
-        input_dir = os.path.join(os.path.dirname(__file__), "input")
-        input_files = [
-            {
-                "filename": "depth_map.png",
-                "type": "images",
-                "url": os.path.join(input_dir, "depth_map.png")
-            },
-            {
-                "filename": "background_bokeh_img2.png",
-                "type": "images",
-                "url": os.path.join(input_dir, "background_bokeh_img2.png")
-            },
-            {
-                "filename": "character_ref_back.png",
-                "type": "images",
-                "url": os.path.join(input_dir, "character_ref_back.png")
-            },
-            {
-                "filename": "easel_blank_canvas_img4.png",
-                "type": "images",
-                "url": os.path.join(input_dir, "easel_blank_canvas_img4.png")
-            }
-        ]
-
-        exec_res = call_tool("runcomfy_workflow_executor", {
-            "server_id": server_id,
-            "workflow_api": workflow_api,
-            "input_files": input_files 
-        })
-        
-        print("Execution finished! Prompt ID:", exec_res.get("prompt_id"))
-        
-        images = exec_res.get("images", [])
-        print("Generated images count:", len(images))
-        if images:
-            # 拿到的是 base64 或者是 CDN url
-            img_data = images[0]
-            if isinstance(img_data, dict):
-                print("First image info:", img_data.get("url") or (img_data.get("data", "")[:50] + "..."))
-            else:
-                print("First image info (raw):", img_data[:50] + "...")
-                
-                # 这是一个 base64 编码的 PNG，我们直接把它保存到 out/ 目录
-                import base64
-                output_dir = os.path.join(os.path.dirname(__file__), "output")
-                os.makedirs(output_dir, exist_ok=True)
-                output_path = os.path.join(output_dir, "test_output.png")
-                
-                try:
-                    with open(output_path, "wb") as fh:
-                        fh.write(base64.b64decode(img_data))
-                    print(f"🎉 成功！生成的图片已保存至: {output_path}")
-                except Exception as e:
-                    print(f"Failed to save image: {e}")
-        
-    finally:
-        # 4. Cleanup
-        print("\n4. Keep machine alive (Skip cleanup so you can check runcomfy)")
-        # 暂时跳过 stop 方便排查
-        # call_tool("runcomfy_stop_env", {"server_id": server_id})
-
-if __name__ == "__main__":
-    test_full_workflow()

tests/fixtures/t2.jpg

@@ -1 +0,0 @@
-test2

tests/fixtures/test.jpg

@@ -1 +0,0 @@
-test image data

tests/liblibai_comfyui_runner.py

@@ -1,175 +0,0 @@
-#!/usr/bin/env python3
-"""LibLibAI ComfyUI Workflow Runner
-
-用法:
-    python liblibai_comfyui_runner.py workflow.json
-    python liblibai_comfyui_runner.py workflow.json --output result.png
-"""
-
-import argparse
-import base64
-import hashlib
-import hmac
-import json
-import os
-import sys
-import time
-import uuid
-from pathlib import Path
-
-import requests
-from dotenv import load_dotenv
-
-load_dotenv()
-
-DOMAIN = os.getenv("LIBLIBAI_DOMAIN", "https://openapi.liblibai.cloud")
-ACCESS_KEY = os.getenv("LIBLIBAI_ACCESS_KEY")
-SECRET_KEY = os.getenv("LIBLIBAI_SECRET_KEY")
-DEFAULT_TEMPLATE = "4df2efa0f18d46dc9758803e478eb51c"
-
-
-def generate_auth_url(uri: str) -> str:
-    """生成带 HMAC-SHA1 签名的完整 URL"""
-    ts = str(int(time.time() * 1000))
-    nonce = uuid.uuid4().hex
-    sign_str = f"{uri}&{ts}&{nonce}"
-    dig = hmac.new(SECRET_KEY.encode(), sign_str.encode(), hashlib.sha1).digest()
-    signature = base64.urlsafe_b64encode(dig).rstrip(b"=").decode()
-    return f"{DOMAIN}{uri}?AccessKey={ACCESS_KEY}&Timestamp={ts}&SignatureNonce={nonce}&Signature={signature}"
-
-
-def submit_workflow(workflow_data: dict) -> str:
-    """提交 ComfyUI workflow 生图任务"""
-    uri = "/api/generate/comfyui/app"
-    url = generate_auth_url(uri)
-
-    payload = {
-        "templateUuid": DEFAULT_TEMPLATE,
-        "generateParams": workflow_data
-    }
-
-    print(f"提交任务...")
-    resp = requests.post(url, json=payload, headers={"Content-Type": "application/json"})
-    resp.raise_for_status()
-
-    data = resp.json()
-    if data.get("code") != 0:
-        raise Exception(f"提交失败: {data.get('msg', 'unknown error')}")
-
-    generate_uuid = data["data"]["generateUuid"]
-    print(f"任务 ID: {generate_uuid}")
-    return generate_uuid
-
-
-def poll_result(generate_uuid: str, timeout: int = 600) -> dict:
-    """轮询任务结果"""
-    uri = "/api/generate/comfy/status"
-    url = generate_auth_url(uri)
-
-    start_time = time.time()
-    print(f"轮询结果 (超时 {timeout}s)...")
-
-    while time.time() - start_time < timeout:
-        resp = requests.post(url, json={"generateUuid": generate_uuid})
-        resp.raise_for_status()
-
-        data = resp.json()
-        if data.get("code") != 0:
-            raise Exception(f"查询失败: {data.get('msg', 'unknown error')}")
-
-        result = data["data"]
-        status = result["generateStatus"]
-
-        # 1=等待 2=执行中 3=已生图 4=审核中 5=成功 6=失败
-        if status in [1, 2, 3, 4]:
-            status_text = {1: "等待", 2: "执行中", 3: "已生图", 4: "审核中"}[status]
-            print(f"  状态: {status_text}  进度: {result.get('percentCompleted', 0):.0%}")
-            time.sleep(5)
-            continue
-
-        if status == 5:
-            print(f"✓ 任务成功")
-            print(f"  消耗积分: {result.get('pointsCost', 0)}")
-            print(f"  剩余积分: {result.get('accountBalance', 0)}")
-            return result
-
-        if status == 6:
-            raise Exception(f"任务失败: {result.get('generateMsg', 'unknown')}")
-
-    raise TimeoutError(f"轮询超时 ({timeout}s)")
-
-
-def save_results(result: dict, output_dir: Path):
-    """保存生成的图片和视频"""
-    output_dir.mkdir(parents=True, exist_ok=True)
-
-    images = result.get("images", [])
-    videos = result.get("videos", [])
-
-    print(f"\n生成结果:")
-    print(f"  图片: {len(images)} 张")
-    print(f"  视频: {len(videos)} 个")
-
-    for i, img in enumerate(images):
-        if img.get("auditStatus") != 3:
-            print(f"  图片 {i+1}: 审核未通过 (status={img.get('auditStatus')})")
-            continue
-
-        url = img["imageUrl"]
-        filename = f"image_{i+1}.png"
-        filepath = output_dir / filename
-
-        print(f"  下载: {filename}")
-        resp = requests.get(url)
-        resp.raise_for_status()
-        filepath.write_bytes(resp.content)
-        print(f"    -> {filepath}")
-
-    for i, vid in enumerate(videos):
-        if vid.get("auditStatus") != 3:
-            print(f"  视频 {i+1}: 审核未通过 (status={vid.get('auditStatus')})")
-            continue
-
-        url = vid["videoUrl"]
-        filename = f"video_{i+1}.mp4"
-        filepath = output_dir / filename
-
-        print(f"  下载: {filename}")
-        resp = requests.get(url)
-        resp.raise_for_status()
-        filepath.write_bytes(resp.content)
-        print(f"    -> {filepath}")
-
-
-def main():
-    parser = argparse.ArgumentParser(description="LibLibAI ComfyUI Workflow Runner")
-    parser.add_argument("workflow", help="workflow JSON 文件路径")
-    parser.add_argument("--output", "-o", default="output", help="输出目录，默认 output/")
-    parser.add_argument("--timeout", "-t", type=int, default=600, help="轮询超时秒数，默认 600")
-    args = parser.parse_args()
-
-    if not ACCESS_KEY or not SECRET_KEY:
-        print("ERROR: 请设置环境变量 LIBLIBAI_ACCESS_KEY 和 LIBLIBAI_SECRET_KEY")
-        sys.exit(1)
-
-    workflow_path = Path(args.workflow)
-    if not workflow_path.exists():
-        print(f"ERROR: 文件不存在: {workflow_path}")
-        sys.exit(1)
-
-    print(f"加载 workflow: {workflow_path}")
-    with open(workflow_path, "r", encoding="utf-8") as f:
-        workflow_data = json.load(f)
-
-    try:
-        generate_uuid = submit_workflow(workflow_data)
-        result = poll_result(generate_uuid, timeout=args.timeout)
-        save_results(result, Path(args.output))
-        print("\n✓ 完成")
-    except Exception as e:
-        print(f"\n✗ 错误: {e}")
-        sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

tests/tasks/image_stitcher.json

@@ -1,4 +0,0 @@
-{
-  "description": "图片拼接工具：将多张图片按指定方式拼接成一张大图",
-  "task_spec": "请创建一个「图片拼接工具」，使用本地 uv 环境。\n\n## 功能需求\n将多张图片按指定方式拼接成一张大图。\n\n## 输入参数（POST /stitch，JSON Body）\n- images: list[str] — Base64 编码的图片列表（至少 2 张）\n- direction: str — 拼接方向，可选 \"horizontal\" | \"vertical\" | \"grid\"，默认 \"horizontal\"\n- columns: int — grid 模式下每行列数，默认 2\n- spacing: int — 图片间距（像素），默认 0\n- background_color: str — 间距填充色，默认 \"#FFFFFF\"\n- resize_mode: str — \"none\" 不缩放 | \"fit_width\" 统一宽度 | \"fit_height\" 统一高度，默认 \"none\"\n\n## 输出（JSON）\n- image: str — 拼接结果，Base64 编码的 PNG\n- width: int — 结果图宽度\n- height: int — 结果图高度\n\n## 技术要求\n1. 使用 uv 环境，项目名 image_stitcher\n2. 核心依赖：Pillow\n3. HTTP 接口：FastAPI + uvicorn，端口通过 --port 参数指定\n4. 路由：POST /stitch（拼接）、GET /health（健康检查，返回 {\"status\":\"ok\"}）\n5. 编写自测脚本 tests/test_stitch.py：生成纯色小图 -> 调用拼接函数 -> 验证输出尺寸，测试产物保存到 tests/output/\n6. 自测通过后注册，tool_id = \"image_stitcher\"，runtime_type = \"local\"，host_dir 填项目目录绝对路径\n\n## 注意\n- 这是 uv 本地项目，不需要 Docker\n- 测试阶段只验证核心逻辑（单元测试），不要启动 HTTP 服务器\n- 先跑通自测脚本再注册，确保核心逻辑正确"
-}

tests/tasks/runcomfy_check_workflow.json

@@ -1,4 +0,0 @@
-{
-  "description": "ComfyUI workflow 输入检查工具：分析 workflow_api.json 所需的外部文件，检查 input 目录是否齐全",
-  "task_spec": "脚本位置：tests/run_comfy/check_workflow.py\n\n## 功能\n分析 workflow_api.json 中所有需要外部文件输入的节点，列出缺失文件，并检查节点连接完整性。\n\n## 用法\n```bash\npython check_workflow.py workflow_api.json\npython check_workflow.py workflow_api.json --input-dir ./my_input\n```\n\n## 参数\n- workflow（位置参数）：workflow_api.json 路径\n- --input-dir：本地输入文件目录，默认 input/，递归扫描所有子目录\n\n## 检查内容\n1. 文件输入节点（LoadImage、LoadVideo、LoadImageMask 等）所需的文件是否存在于 input 目录\n2. 节点连接是否完整（引用的 src_node_id 是否存在）\n3. 是否存在 widget_* 占位参数（参数名不准确的警告）\n\n## 输出\n- ✓ 检查通过：所有文件齐全，连接完整\n- ❌ 检查未通过：列出缺失文件和连接问题\n- ⚠️ 警告：存在 widget_* 占位参数\n\n## 支持的文件输入节点类型\n- LoadImage → image 参数\n- LoadVideo / LoadVideoPath / VHS_LoadVideo → video 参数\n- LoadImageMask → image 参数\n- LoadImageOutput → image 参数（从 output 目录加载，特殊情况）\n- VHS_LoadImages → directory 参数"
-}

tests/tasks/runcomfy_convert_workflow.json

@@ -1,4 +0,0 @@
-{
-  "description": "ComfyUI workflow 格式转换工具：将 ComfyUI UI 导出的 workflow.json 转换为可通过 API 提交的 workflow_api.json",
-  "task_spec": "脚本位置：tests/run_comfy/convert_workflow.py\n\n## 功能\n将 ComfyUI 界面导出的 workflow.json（UI格式）转换为 API 可直接提交的 workflow_api.json（API格式）。\n\n## 两种格式的区别\n- UI格式：nodes[] 数组，每个节点含 id/type/pos/size/color/widgets_values/links 等 UI 信息\n- API格式：以 node_id 为 key 的对象，每个节点只含 class_type 和 inputs（参数值或连接引用）\n\n## 用法\n```bash\npython convert_workflow.py workflow.json\npython convert_workflow.py workflow.json -o output_api.json\n```\n\n## 参数\n- workflow（位置参数）：输入的 UI 格式 workflow.json 路径\n- -o / --output：输出路径，默认在原文件名基础上加 _api 后缀\n\n## 转换规则\n1. 跳过纯 UI 节点（Note、MarkdownNote、PrimitiveNode）\n2. 跳过 mode != 0 的禁用节点\n3. 节点连接通过 links[] 映射还原为 [src_node_id, src_slot] 格式\n4. widgets_values 按顺序填入未连线的输入参数，参数名用 widget_0/widget_1/... 占位\n\n## 注意\n- widget_* 占位参数名不准确，建议直接在 ComfyUI 界面使用「Save (API Format)」按钮导出，可获得准确参数名\n- 转换后建议用 check_workflow.py 验证"
-}

tests/tasks/runcomfy_stop_env.json

@@ -1,7 +0,0 @@
-{
-  "description": "RunComfy 环境销毁工具：根据 server_id 主动关闭已启动的 ComfyUI 机器，避免持续计费",
-  "task_spec": "## 目标\n创建一个 HTTP API 工具并注册到 Router，用于关闭 RunComfy 机器实例。\n\n## 核心功能\n调用 RunComfy Server API 删除指定的机器实例，释放资源。配合 `launch_comfy_env` 使用，负责生命周期的收尾阶段。\n\n## 环境变量\n- RUNCOMFY_USER_ID：RunComfy 用户 ID\n- API_TOKEN：RunComfy API Token\n\n## HTTP API 接口（必须实现）\n实现 `POST /stop` 接口：\n\n### 输入 JSON Schema\n```json\n{\n  \"type\": \"object\",\n  \"properties\": {\n    \"server_id\": {\"type\": \"string\", \"description\": \"要关闭的机器 ID\"}\n  },\n  \"required\": [\"server_id\"]\n}\n```\n\n### 输出 JSON Schema\n```json\n{\n  \"type\": \"object\",\n  \"properties\": {\n    \"server_id\": {\"type\": \"string\", \"description\": \"关闭的机器 ID\"},\n    \"status\": {\"type\": \"string\", \"description\": \"Deleted 或报错信息\"},\n    \"message\": {\"type\": \"string\", \"description\": \"详细结果文本\"}\n  }\n}\n```\n\n## 核心逻辑\n1. 调用 API `DELETE https://beta-api.runcomfy.net/prod/api/users/{USER_ID}/servers/{server_id}`\n2. Header 需要 `Authorization: Bearer {API_TOKEN}`\n3. 处理 200 响应和 404 响应（说明已被清理或不存在），如果是其它错误，则抛出异常并返回失败状态\n4. 这个工具应始终返回安全的明确状态，供 Agent 确认释放情况\n\n## 实现要求\n1. 使用 uv 创建项目，项目名：runcomfy_stop_env\n2. 使用 FastAPI 实现 HTTP 接口\n3. 从环境变量读取 RUNCOMFY_USER_ID 和 API_TOKEN\n4. 实现 POST /stop 接口\n5. 编写测试脚本验证功能\n6. **必须调用 register_tool 注册到 Router**，tool_id 为 \"runcomfy_stop_env\"",
-  "reference_files": [
-    "tests/run_comfy/run_workflow.py"
-  ]
-}

tests/test_chat.py

@@ -1,11 +0,0 @@
-import asyncio
-import httpx
-import uuid
-
-async def test():
-    async with httpx.AsyncClient() as client:
-        resp = await client.post("http://127.0.0.1:8001/chat", json={"message": "列举你的tool", "chat_id": uuid.uuid4().hex})
-        print(resp.status_code)
-        print(resp.text)
-
-asyncio.run(test())

tests/test_create_runcomfy_atomic.py

@@ -1,94 +0,0 @@
-"""批量触发生成三个 RunComfy 原子化工具
-
-用法：
-    uv run python tests/test_create_runcomfy_atomic.py
-"""
-
-import sys
-import time
-from pathlib import Path
-
-import httpx
-
-BASE_URL = "http://127.0.0.1:8001"
-TASKS_DIR = Path(__file__).parent / "tasks"
-
-# 我们刚才写的三个原子化任务书
-TASKS = [
-    "runcomfy_launch_env",
-    "runcomfy_run_only",
-    "runcomfy_stop_env"
-]
-
-def check_connection():
-    try:
-        httpx.get(f"{BASE_URL}/health", timeout=3)
-    except httpx.ConnectError:
-        print(f"ERROR: Cannot connect to {BASE_URL}")
-        print("Please start the service first:")
-        print("  uv run python -m tool_agent")
-        sys.exit(1)
-
-def submit_task(task_name: str) -> str:
-    task_file = TASKS_DIR / f"{task_name}.json"
-    if not task_file.exists():
-        print(f"ERROR: Task file not found: {task_file}")
-        sys.exit(1)
-
-    import json
-    with open(task_file, "r", encoding="utf-8") as f:
-        task_data = json.load(f)
-
-    print(f"\n[{task_name}] Submitting...")
-    resp = httpx.post(f"{BASE_URL}/create_tool", json=task_data, timeout=30)
-    resp.raise_for_status()
-    data = resp.json()
-    task_id = data["task_id"]
-    print(f"[{task_name}] Task ID: {task_id}")
-    return task_id
-
-def poll_tasks(task_ids: dict[str, str], timeout: int = 900):
-    print("\n=== Polling Tasks ===")
-    pending = set(task_ids.values())
-
-    interval = 10
-    steps = timeout // interval
-
-    for i in range(steps):
-        if not pending:
-            print("\nAll tasks finished!")
-            break
-
-        time.sleep(interval)
-        elapsed = (i + 1) * interval
-
-        for task_name, task_id in list(task_ids.items()):
-            if task_id not in pending:
-                continue
-
-            resp = httpx.get(f"{BASE_URL}/tasks/{task_id}", timeout=30)
-            status = resp.json()["status"]
-
-            if status in ("completed", "failed"):
-                print(f"\n[{task_name}] Finished with status: {status}")
-                pending.remove(task_id)
-            elif elapsed % 30 == 0:
-                print(f"[{elapsed}s] {task_name}: {status}")
-
-    if pending:
-        print(f"\nTimeout! Still pending: {pending}")
-
-def main():
-    check_connection()
-
-    # 1. 批量提交
-    task_ids = {}
-    for task_name in TASKS:
-        task_id = submit_task(task_name)
-        task_ids[task_name] = task_id
-
-    # 2. 并行轮询
-    poll_tasks(task_ids)
-
-if __name__ == "__main__":
-    main()

tests/test_dispatcher.py

@@ -1,10 +0,0 @@
-"""测试 Dispatcher"""
-
-import pytest
-from tool_agent.router.dispatcher import Dispatcher
-
-
-class TestDispatcher:
-    def test_dispatcher_init(self):
-        d = Dispatcher()
-        assert d is not None

tests/test_docker_git.py

@@ -1,97 +0,0 @@
-"""测试 DockerRunner: 创建容器 → 安装配置 git → clone 仓库 → 清理"""
-
-import sys
-import logging
-
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s")
-logger = logging.getLogger("test_docker_git")
-
-
-def main():
-    from tool_agent.runtime.docker_runner import DockerRunner
-
-    runner = DockerRunner(lazy_init=True)
-
-    # ---- Step 1: 创建容器 ----
-    logger.info("=" * 60)
-    logger.info("Step 1: 创建容器 (ubuntu:22.04)")
-    result = runner.create_container(
-        tool_id="test_git_tool",
-        image="ubuntu:22.04",
-        mem_limit="512m",
-        nano_cpus=1_000_000_000,
-    )
-
-    if "error" in result:
-        logger.error(f"创建容器失败: {result['error']}")
-        sys.exit(1)
-
-    container_id = result["container_id"]
-    logger.info(f"容器已创建: {container_id[:12]}")
-
-    try:
-        # ---- Step 2: 安装 git ----
-        logger.info("=" * 60)
-        logger.info("Step 2: 安装 git")
-        res = runner.run_command(
-            container_id,
-            "apt-get update -qq && apt-get install -y -qq git > /dev/null 2>&1 && git --version",
-            timeout=120,
-        )
-        if res.get("exit_code", -1) != 0:
-            logger.error(f"安装 git 失败: {res}")
-            sys.exit(1)
-        logger.info(f"git 安装成功: {res['stdout'].strip()}")
-
-        # ---- Step 3: 配置 git ----
-        logger.info("=" * 60)
-        logger.info("Step 3: 配置 git 用户")
-        res = runner.run_command(
-            container_id,
-            'git config --global user.name "Tool Agent" && '
-            'git config --global user.email "agent@tool-agent.local" && '
-            'git config --global --list',
-        )
-        if res.get("exit_code", -1) != 0:
-            logger.error(f"配置 git 失败: {res}")
-            sys.exit(1)
-        logger.info(f"git 配置:\n{res['stdout'].strip()}")
-
-        # ---- Step 4: clone 仓库 ----
-        logger.info("=" * 60)
-        logger.info("Step 4: git clone 仓库")
-        res = runner.run_command(
-            container_id,
-            "cd /app && git clone --depth 1 https://github.com/pallets/flask.git",
-            timeout=120,
-        )
-        if res.get("exit_code", -1) != 0:
-            logger.error(f"clone 失败: {res}")
-            sys.exit(1)
-        logger.info("clone 完成")
-
-        # ---- Step 5: 验证 clone 结果 ----
-        logger.info("=" * 60)
-        logger.info("Step 5: 验证 clone 结果")
-        res = runner.run_command(
-            container_id,
-            "ls -la /app/flask/ && echo '---' && cd /app/flask && git log --oneline -3",
-        )
-        if res.get("exit_code", -1) != 0:
-            logger.error(f"验证失败: {res}")
-            sys.exit(1)
-        logger.info(f"仓库内容:\n{res['stdout'].strip()}")
-
-        logger.info("=" * 60)
-        logger.info("ALL TESTS PASSED!")
-
-    finally:
-        # ---- Step 6: 清理 ----
-        logger.info("=" * 60)
-        logger.info("Step 6: 销毁容器")
-        cleanup = runner.destroy_container(container_id)
-        logger.info(f"清理结果: {cleanup}")
-
-
-if __name__ == "__main__":
-    main()

tests/test_e2e_stitcher.py

@@ -1,138 +0,0 @@
-"""端到端测试：CodingAgent 自主完成图片拼接工具的编写、测试和注册
-
-运行方式：
-    uv run python -m tests.test_e2e_stitcher
-
-前置条件：
-    - ANTHROPIC_API_KEY 已设置
-    - uv 已安装
-"""
-
-import asyncio
-import json
-import logging
-import sys
-from pathlib import Path
-
-# 确保项目根目录在 sys.path 中
-ROOT = Path(__file__).resolve().parent.parent
-sys.path.insert(0, str(ROOT / "src"))
-
-from tool_agent.config import settings
-from tool_agent.tool.agent import CodingAgent
-from tool_agent.registry.registry import ToolRegistry
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
-)
-logger = logging.getLogger("e2e_test")
-
-TOOL_ID = "image_stitcher"
-
-TASK_SPEC = f"""\
-请创建一个「图片拼接工具」，使用本地 uv 环境。
-
-## 功能需求
-将多张图片按指定方式拼接成一张大图。
-
-## 输入参数（POST /stitch，JSON Body）
-- images: list[str] — Base64 编码的图片列表（至少 2 张）
-- direction: str — 拼接方向，可选 "horizontal" | "vertical" | "grid"，默认 "horizontal"
-- columns: int — grid 模式下每行列数，默认 2
-- spacing: int — 图片间距（像素），默认 0
-- background_color: str — 间距填充色，默认 "#FFFFFF"
-- resize_mode: str — "none" 不缩放 | "fit_width" 统一宽度 | "fit_height" 统一高度，默认 "none"
-
-## 输出（JSON）
-- image: str — 拼接结果，Base64 编码的 PNG
-- width: int — 结果图宽度
-- height: int — 结果图高度
-
-## 技术要求
-1. 使用 uv 环境，项目名 {TOOL_ID}
-2. 核心依赖：Pillow
-3. HTTP 接口：FastAPI + uvicorn，端口自选
-4. 路由：POST /stitch（拼接）、GET /health（健康检查，返回 {{"status":"ok"}}）
-5. 编写一个自测脚本 test_stitch.py：生成几张纯色小图 → 调用拼接函数 → 验证输出尺寸正确
-6. 自测通过后，使用 register_tool 注册，tool_id = "{TOOL_ID}"
-
-## 注意
-- 这是 uv 本地项目，不需要 Docker
-- 先跑通自测脚本再注册，确保核心逻辑正确
-"""
-
-
-async def run_test() -> bool:
-    """执行端到端测试，返回是否通过"""
-
-    logger.info("=" * 60)
-    logger.info("E2E Test: Image Stitcher Tool")
-    logger.info("=" * 60)
-
-    # ---- 1. 执行任务 ----
-    agent = CodingAgent()
-    logger.info("Sending task to CodingAgent...")
-    result = await agent.execute(TASK_SPEC)
-
-    logger.info("Agent finished. Result:")
-    logger.info(result)
-
-    # ---- 2. 验证注册 ----
-    logger.info("-" * 60)
-    logger.info("Verifying registration...")
-
-    registry = ToolRegistry()
-    tool = registry.get(TOOL_ID)
-
-    if not tool:
-        logger.error(f"FAIL: tool '{TOOL_ID}' not found in registry")
-        return False
-
-    logger.info(f"OK: tool registered — {tool.tool_id} ({tool.name})")
-    logger.info(f"  runtime: {tool.runtime.type.value}")
-    logger.info(f"  status:  {tool.status.value}")
-
-    endpoint = registry.get_endpoint(TOOL_ID)
-    if endpoint:
-        logger.info(f"  endpoint: {json.dumps(endpoint, ensure_ascii=False)}")
-    else:
-        logger.warning("  endpoint: None (may be expected for local stdio tool)")
-
-    # ---- 3. 验证项目文件 ----
-    logger.info("-" * 60)
-    logger.info("Verifying project files...")
-
-    project_dir = settings.tools_dir / "local" / TOOL_ID
-    checks = {
-        "project dir exists": project_dir.is_dir(),
-        "pyproject.toml":     (project_dir / "pyproject.toml").exists(),
-    }
-
-    all_pass = True
-    for label, ok in checks.items():
-        status = "OK" if ok else "FAIL"
-        logger.info(f"  {status}: {label}")
-        if not ok:
-            all_pass = False
-
-    # 列出项目内文件
-    if project_dir.is_dir():
-        files = sorted(p.relative_to(project_dir) for p in project_dir.rglob("*") if p.is_file())
-        logger.info(f"  project files ({len(files)}):")
-        for f in files:
-            logger.info(f"    {f}")
-
-    # ---- 4. 汇总 ----
-    logger.info("=" * 60)
-    if all_pass and tool:
-        logger.info("E2E TEST PASSED")
-    else:
-        logger.error("E2E TEST FAILED")
-
-    return all_pass and tool is not None
-
-
-if __name__ == "__main__":
-    ok = asyncio.run(run_test())
-    sys.exit(0 if ok else 1)

tests/test_extrct/extract_atomic_capabilities.prompt

@@ -1,52 +0,0 @@
----
-name: extract_atomic_capabilities
-model: anthropic/claude-sonnet-4.6
-temperature: 0.3
-max_tokens: 16000
----
-
-$system$
-你是一个专业的能力分析师。你的任务是从工具的使用介绍和实际用例中提取**原子能力**。
-
-## 什么是原子能力？
-
-原子能力是一种**面向需求的、跨工具的高维能力**。它不是某个工具的具体技术实现细节，而是一种独立完整的、可直接面对用户需求的能力单元。
-
-### 核心定义
-- 它是**面向需求**的：每个原子能力都直接对应用户的某一类创作需求
-- 它是**跨工具**的：同一个原子能力可以由不同工具以不同方式实现
-- 它是**不可分割**的：拆分后将无法独立满足任何需求
-- 它是**可组合**的：多个原子能力按顺序组合可形成完整的工序/流水线
-
-### 关于工具的自由度差异
-
-工具分为两类，提取原子能力时请注意区分：
-
-**端到端工具**（如 Midjourney、DALL-E）：输入 prompt → 输出图像，能力边界清晰。
-→ 从其参数/功能中直接提取原子能力。
-
-**编排平台型工具**（如 ComfyUI、HTML）：内部高度自由，可任意组合节点/模块，能力边界开放。
-→ 不要试图原子化平台本身，而是从其**具体的工作流和用例**中提取原子能力。
-→ 在「实现方式」中标注具体的工作流/方案，而非泛泛地写"ComfyUI"。
-→ 例如：ComfyUI 实现「角色一致性」的方式是「IP-Adapter 节点 + 参考图」，这就是一种实现方案。
-
-### 举例说明
-✅ **正确的原子能力**：「保持角色一致性」— 跨多张图保持同一角色的面部/身体/服装特征不变。这个能力可由 ControlNet、IP-Adapter、--cref 参数、多图参考等不同工具/方式实现，但核心需求是一样的。
-✅ **正确的原子能力**：「图内文字渲染」— 在生成的图像中嵌入清晰可读的指定文字。
-❌ **错误（太底层）**：「使用 KSampler 采样」— 这是具体技术操作，不是面向需求的能力。
-❌ **错误（太底层）**：「设置 --ar 16:9」— 这是参数设置，不是独立能力。
-❌ **错误（可再分）**：「制作电商产品图」— 这可以分解为「背景替换」+「产品一致性保持」+「光照调整」等多个原子能力的组合。
-
-## 原子能力的格式
-
-每个原子能力应包含：
-
-### [能力ID]: [能力名称]
-- **功能描述**: [做什么，满足什么需求]
-- **判定标准**: [怎样算做到了，怎样算没做到]
-- **实现方式**: [列举可实现的工具/方案。端到端工具直接写工具名+参数；编排平台写具体工作流，如「ComfyUI: IP-Adapter节点+参考图输入」]
-- **典型场景**: [什么时候需要用到这个能力]
-- **来源依据**: [从哪些具体用例/文档提炼出来的，简述来源帖子或用例的大概内容]
-
-$user$
-{user_prompt}

tests/test_extrct/extract_capabilities_auto.py

@@ -1,325 +0,0 @@
-#!/usr/bin/env python3
-"""
-原子能力提取工作流 - 自动化版本
-使用 openrouter (claude-sonnet) 逐个读取工具文档，迭代式提取和融合原子能力
-"""
-
-import asyncio
-import json
-import os
-import re
-import sys
-from pathlib import Path
-
-# 添加项目根目录
-sys.path.insert(0, str(Path(__file__).parent.parent.parent))
-
-from dotenv import load_dotenv
-load_dotenv()
-
-from agent.llm.openrouter import openrouter_llm_call
-
-# ===== 配置 =====
-BASE_DIR = Path(__file__).parent
-TOOL_RESULTS_DIR = BASE_DIR / "tool_results"
-OUTPUT_FILE = BASE_DIR / "atomic_capabilities.md"
-PROMPT_FILE = BASE_DIR / "extract_atomic_capabilities.prompt"
-
-
-# ===== Prompt 加载（复用 match_nodes.py 的模式） =====
-
-def load_prompt(filepath: str) -> dict:
-    """加载 .prompt 文件，解析 frontmatter 和 $role$ 分段"""
-    text = Path(filepath).read_text(encoding="utf-8")
-
-    config = {}
-    if text.startswith("---"):
-        _, fm, text = text.split("---", 2)
-        for line in fm.strip().splitlines():
-            if ":" in line:
-                k, v = line.split(":", 1)
-                k, v = k.strip(), v.strip()
-                if v.replace(".", "", 1).isdigit():
-                    v = float(v) if "." in v else int(v)
-                config[k] = v
-
-    messages = []
-    parts = re.split(r'^\$(\w+)\$\s*$', text.strip(), flags=re.MULTILINE)
-    for i in range(1, len(parts), 2):
-        role = parts[i].strip()
-        content = parts[i + 1].strip() if i + 1 < len(parts) else ""
-        messages.append({"role": role, "content": content})
-
-    return {"config": config, "messages": messages}
-
-
-def render_messages(prompt_data: dict, variables: dict) -> list[dict]:
-    """用变量替换 prompt 模板中的 {var} 占位符"""
-    rendered = []
-    for msg in prompt_data["messages"]:
-        content = msg["content"]
-        for k, v in variables.items():
-            content = content.replace(f"{{{k}}}", str(v))
-        rendered.append({"role": msg["role"], "content": content})
-    return rendered
-
-
-# ===== 文件读取 =====
-
-def get_all_tool_dirs():
-    """获取所有工具目录"""
-    dirs = sorted([d for d in TOOL_RESULTS_DIR.iterdir() if d.is_dir()])
-    return dirs
-
-
-def read_file(file_path):
-    """读取文件内容"""
-    with open(file_path, 'r', encoding='utf-8') as f:
-        return f.read()
-
-
-def read_tool_files(tool_dir):
-    """读取工具的使用介绍和实际用例"""
-    usage_file = tool_dir / "使用介绍.md"
-    case_file = tool_dir / "实际用例.md"
-
-    content = ""
-    if usage_file.exists():
-        content += "# 使用介绍\n\n" + read_file(usage_file) + "\n\n"
-    if case_file.exists():
-        content += "# 实际用例\n\n" + read_file(case_file)
-
-    return content
-
-
-# ===== 构建 user prompt =====
-
-def build_user_prompt(file_content, tool_name, existing_capabilities=""):
-    """构建每轮迭代的 user prompt"""
-
-    if existing_capabilities:
-        user_prompt = f"""## 当前状态
-
-### 已提取的原子能力
-
-{existing_capabilities}
-
-## 你的工作
-
-1. 仔细阅读下面的工具文档（使用介绍 + 实际用例）
-2. 从中识别出**面向需求的原子能力**（注意：不是工具的技术操作）
-3. 与已有能力对比：
-   - 如果是全新的能力 → 添加，并说明来源
-   - 如果已有能力可由新工具实现 → 融合，在「实现方式」中补充该工具
-   - 如果是多个已有能力的组合 → 不添加，但在「发现的能力组合」中记录
-4. 对于来源依据，要说明从哪个用例/帖子/文档章节提炼的，并简述其大概内容
-"""
-    else:
-        user_prompt = """## 当前状态
-
-这是第一次提取，当前没有已有能力。
-
-## 你的工作
-
-1. 仔细阅读下面的工具文档（使用介绍 + 实际用例）
-2. 从中识别出**面向需求的原子能力**（注意：不是工具的技术操作）
-3. 对于来源依据，要说明从哪个用例/帖子/文档章节提炼的，并简述其大概内容
-"""
-
-    user_prompt += f"""
-## 当前要处理的工具
-
-**工具名称**: {tool_name}
-
-**文档内容**（包含使用介绍和实际用例）:
-
-{file_content}
-
-## 输出要求
-
-请按以下格式输出：
-
-# 原子能力清单（更新后）
-
-## 本轮分析
-简要说明从 {tool_name} 中发现了哪些能力，哪些是新的，哪些与已有能力融合了。
-
-## 新增能力
-[列出本次新增的能力，使用上述格式，每个能力都要有来源依据]
-
-## 融合能力
-[列出本次融合/更新的能力，说明新增了哪些实现方式]
-
-## 发现的能力组合
-[列出发现的能力组合关系，例如：能力A + 能力B + 能力C = 完成「电商产品图批量生成」]
-
-## 完整能力清单
-[输出完整的、更新后的原子能力清单，包含所有能力（新增 + 已有 + 融合后的）]
-"""
-
-    return user_prompt
-
-
-# ===== LLM 调用 =====
-
-async def extract_capabilities_from_tool(prompt_data, tool_dir, existing_capabilities=""):
-    """从工具目录提取原子能力"""
-    tool_name = tool_dir.name
-    print(f"\n📖 正在处理: {tool_name}")
-
-    # 读取使用介绍和实际用例
-    content = read_tool_files(tool_dir)
-
-    # 构建 user prompt
-    user_prompt = build_user_prompt(content, tool_name, existing_capabilities)
-
-    # 渲染 prompt 模板
-    messages = render_messages(prompt_data, {"user_prompt": user_prompt})
-
-    # 从 prompt 文件读取配置
-    model = prompt_data["config"].get("model", "anthropic/claude-sonnet-4-20250514")
-    temperature = prompt_data["config"].get("temperature", 0.3)
-    max_tokens = prompt_data["config"].get("max_tokens", 16000)
-
-    try:
-        result = await openrouter_llm_call(
-            messages, model=model, temperature=temperature, max_tokens=max_tokens
-        )
-        response = result["content"]
-
-        # 打印 token 用量
-        pt = result.get("prompt_tokens", 0)
-        ct = result.get("completion_tokens", 0)
-        cost = result.get("cost", 0)
-        print(f"   tokens: {pt} prompt + {ct} completion | cost: ${cost:.4f}")
-
-        # 提取"完整能力清单"部分
-        if "## 完整能力清单" in response:
-            complete_list = response.split("## 完整能力清单")[1].strip()
-        else:
-            complete_list = response
-
-        print(f"✅ {tool_name} 处理完成")
-        return response, complete_list
-
-    except Exception as e:
-        print(f"❌ {tool_name} 处理失败: {e}")
-        return None, existing_capabilities
-
-
-async def generate_json_index(prompt_data, capabilities_md):
-    """把 markdown 格式的能力清单转成简洁 JSON"""
-    prompt = f"""请把以下原子能力清单转成 JSON 数组，每个能力包含以下字段：
-
-```json
-[
-  {{
-    "id": "能力ID",
-    "name": "能力名称",
-    "description": "一句话功能描述",
-    "criteria": "判定标准（简洁）",
-    "tools": ["支持的工具/方案1", "支持的工具/方案2"],
-    "scenarios": ["典型场景1", "典型场景2"],
-    "source_summary": "来源依据的简要概括"
-  }}
-]
-```
-
-要求：
-- 只输出 JSON，不要任何其他文字
-- 保持所有能力，不要遗漏
-- description 控制在 30 字以内
-- criteria 控制在 30 字以内
-
-原子能力清单：
-
-{capabilities_md}
-"""
-
-    model = prompt_data["config"].get("model", "anthropic/claude-sonnet-4-20250514")
-
-    messages = [{"role": "user", "content": prompt}]
-    try:
-        result = await openrouter_llm_call(messages, model=model, temperature=0.1, max_tokens=8000)
-        content = result["content"].strip()
-        # 清理 markdown 代码块包裹
-        if content.startswith("```"):
-            content = content.split("\n", 1)[1]
-            content = content.rsplit("```", 1)[0]
-        # 验证是合法 JSON
-        json.loads(content)
-        return content
-    except Exception as e:
-        print(f"❌ JSON 索引生成失败: {e}")
-        return None
-
-
-# ===== 主流程 =====
-
-async def main():
-    print("🚀 开始提取原子能力...")
-    print()
-
-    # 加载 prompt 模板
-    prompt_data = load_prompt(PROMPT_FILE)
-    model = prompt_data["config"].get("model", "anthropic/claude-sonnet-4-20250514")
-    print(f"🤖 使用模型: {model}")
-    print()
-
-    # 获取所有工具目录
-    tool_dirs = get_all_tool_dirs()
-    print(f"📁 找到 {len(tool_dirs)} 个工具:")
-    for d in tool_dirs:
-        files = list(d.glob("*.md"))
-        print(f"   - {d.name} ({len(files)} 个文件)")
-    print()
-
-    # 迭代处理每个工具
-    existing_capabilities = ""
-    all_responses = []
-
-    for i, tool_dir in enumerate(tool_dirs, 1):
-        print(f"{'='*60}")
-        print(f"进度: [{i}/{len(tool_dirs)}]")
-
-        response, complete_list = await extract_capabilities_from_tool(
-            prompt_data, tool_dir, existing_capabilities
-        )
-
-        if response:
-            all_responses.append({
-                "tool": tool_dir.name,
-                "response": response
-            })
-            existing_capabilities = complete_list
-
-    # 保存最终结果
-    print(f"\n{'='*60}")
-    print("💾 保存结果...")
-
-    # 保存完整能力清单（markdown）
-    OUTPUT_FILE.write_text(existing_capabilities, encoding='utf-8')
-    print(f"✅ 原子能力清单已保存到: {OUTPUT_FILE}")
-
-    # 保存详细过程
-    detail_file = BASE_DIR / "atomic_capabilities_detail.json"
-    with open(detail_file, 'w', encoding='utf-8') as f:
-        json.dump(all_responses, f, ensure_ascii=False, indent=2)
-    print(f"✅ 详细过程已保存到: {detail_file}")
-
-    # 最终一轮：让 LLM 把完整能力清单转成简洁 JSON
-    print(f"\n{'='*60}")
-    print("📋 生成简洁 JSON 索引...")
-    json_result = await generate_json_index(prompt_data, existing_capabilities)
-    if json_result:
-        json_index_file = BASE_DIR / "atomic_capabilities_index.json"
-        with open(json_index_file, 'w', encoding='utf-8') as f:
-            f.write(json_result)
-        print(f"✅ JSON 索引已保存到: {json_index_file}")
-
-    print("\n🎉 所有文件处理完成！")
-
-
-if __name__ == "__main__":
-    os.environ.setdefault("no_proxy", "*")
-    asyncio.run(main())

tests/test_flux.py

@@ -1,177 +0,0 @@
-"""测试 BFL FLUX 异步生图 — 通过 Router POST /run_tool
-
-官方流程：先 POST 提交任务拿到 id + polling_url，再轮询 polling_url 直至 Ready。
-文档: https://docs.bfl.ai/quick_start/generating_images
-
-用法：
-    1. 配置 tools/local/flux/.env：BFL_API_KEY
-    2. uv run python -m tool_agent
-    3. uv run python tests/test_flux.py
-
-模型切换：
-    FLUX_TEST_MODEL=flux-2-max uv run python tests/test_flux.py
-    （model 为路径段，如 flux-2-pro-preview、flux-2-pro、flux-dev 等，见官方 Available Endpoints）
-
-环境变量：
-    TOOL_AGENT_ROUTER_URL   默认 http://127.0.0.1:8001
-    FLUX_SUBMIT_TOOL_ID     默认 flux_submit
-    FLUX_QUERY_TOOL_ID      默认 flux_query
-    FLUX_TEST_MODEL         默认 flux-2-pro-preview
-    FLUX_TEST_PROMPT        覆盖默认短提示词
-    FLUX_POLL_INTERVAL_S    默认 1.0
-    FLUX_POLL_MAX_WAIT_S    默认 300
-"""
-
-from __future__ import annotations
-
-import io
-import os
-import sys
-import time
-from typing import Any
-
-if sys.platform == "win32":
-    _out = sys.stdout
-    if isinstance(_out, io.TextIOWrapper):
-        _out.reconfigure(encoding="utf-8")
-
-import httpx
-
-ROUTER_URL = os.environ.get("TOOL_AGENT_ROUTER_URL", "http://127.0.0.1:8001")
-SUBMIT_TOOL = os.environ.get("FLUX_SUBMIT_TOOL_ID", "flux_submit")
-QUERY_TOOL = os.environ.get("FLUX_QUERY_TOOL_ID", "flux_query")
-FLUX_MODEL = os.environ.get("FLUX_TEST_MODEL", "flux-2-pro-preview").strip()
-TEST_PROMPT = os.environ.get(
-    "FLUX_TEST_PROMPT",
-    "A tiny red apple on white background, simple product photo, minimal",
-)
-POLL_INTERVAL_S = float(os.environ.get("FLUX_POLL_INTERVAL_S", "1.0"))
-POLL_MAX_WAIT_S = float(os.environ.get("FLUX_POLL_MAX_WAIT_S", "300"))
-
-
-def run_tool(tool_id: str, params: dict[str, Any], timeout: float = 120.0) -> dict[str, Any]:
-    resp = httpx.post(
-        f"{ROUTER_URL}/run_tool",
-        json={"tool_id": tool_id, "params": params},
-        timeout=timeout,
-    )
-    resp.raise_for_status()
-    body = resp.json()
-    if body.get("status") != "success":
-        raise RuntimeError(body.get("error") or str(body))
-    result = body.get("result")
-    if isinstance(result, dict) and result.get("status") == "error":
-        raise RuntimeError(result.get("error", str(result)))
-    return result if isinstance(result, dict) else {}
-
-
-def _poll_terminal_success(data: dict[str, Any]) -> bool:
-    s = str(data.get("status") or "").strip()
-    return s.lower() == "ready"
-
-
-def _poll_terminal_failure(data: dict[str, Any]) -> bool:
-    s = str(data.get("status") or "").strip().lower()
-    return s in ("error", "failed")
-
-
-def _sample_url(data: dict[str, Any]) -> str | None:
-    r = data.get("result")
-    if isinstance(r, dict):
-        u = r.get("sample")
-        if isinstance(u, str) and u.startswith("http"):
-            return u
-    return None
-
-
-def main() -> None:
-    print("=" * 50)
-    print("测试 FLUX（BFL 异步 API + 模型可切换）")
-    print("=" * 50)
-    print(f"ROUTER_URL: {ROUTER_URL}")
-    print(f"model:      {FLUX_MODEL}")
-
-    try:
-        r = httpx.get(f"{ROUTER_URL}/health", timeout=3)
-        print(f"Router 状态: {r.json()}")
-    except httpx.ConnectError:
-        print(f"无法连接 Router ({ROUTER_URL})，请先: uv run python -m tool_agent")
-        sys.exit(1)
-
-    print("\n--- 校验工具已注册 ---")
-    tr = httpx.get(f"{ROUTER_URL}/tools", timeout=30)
-    tr.raise_for_status()
-    tools = tr.json().get("tools", [])
-    ids = {t["tool_id"] for t in tools}
-    for tid in (SUBMIT_TOOL, QUERY_TOOL):
-        if tid not in ids:
-            print(f"错误: {tid!r} 不在 GET /tools 中。示例 id: {sorted(ids)[:20]}...")
-            sys.exit(1)
-        meta = next(t for t in tools if t["tool_id"] == tid)
-        print(f"  {tid}: {meta.get('name', '')} (state={meta.get('state')})")
-
-    props = (next(t for t in tools if t["tool_id"] == SUBMIT_TOOL).get("input_schema") or {}).get(
-        "properties"
-    ) or {}
-    if "model" in props:
-        print("  flux_submit input_schema 已声明 model")
-    else:
-        print("  提示: flux_submit 宜在注册表中声明 model 以便切换端点")
-
-    print("\n--- flux_submit ---")
-    submit_params: dict[str, Any] = {
-        "model": FLUX_MODEL,
-        "prompt": TEST_PROMPT,
-        "width": 512,
-        "height": 512,
-    }
-    try:
-        sub = run_tool(SUBMIT_TOOL, submit_params, timeout=120.0)
-    except (RuntimeError, httpx.HTTPError) as e:
-        print(f"错误: {e}")
-        sys.exit(1)
-
-    print(f"提交返回 keys: {list(sub.keys())}")
-    req_id = sub.get("id") or sub.get("request_id")
-    poll_url = sub.get("polling_url")
-    if not req_id or not poll_url:
-        print(f"错误: 缺少 id 或 polling_url: {sub}")
-        sys.exit(1)
-    print(f"request id: {req_id}")
-    print(f"polling_url: {poll_url[:80]}...")
-
-    print("\n--- flux_query 轮询 ---")
-    deadline = time.monotonic() + POLL_MAX_WAIT_S
-    last: dict[str, Any] = {}
-
-    while time.monotonic() < deadline:
-        time.sleep(POLL_INTERVAL_S)
-        try:
-            last = run_tool(
-                QUERY_TOOL,
-                {"polling_url": str(poll_url), "request_id": str(req_id)},
-                timeout=60.0,
-            )
-        except (RuntimeError, httpx.HTTPError) as e:
-            print(f"轮询错误: {e}")
-            sys.exit(1)
-
-        st = last.get("status")
-        print(f"  status: {st}")
-
-        if _poll_terminal_failure(last):
-            print(f"生成失败: {last}")
-            sys.exit(1)
-        if _poll_terminal_success(last):
-            url = _sample_url(last)
-            if url:
-                print(f"\n图片 URL（signed，约 10 分钟内有效）: {url[:100]}...")
-            print("\n测试通过!")
-            return
-
-    print(f"\n等待超时 ({POLL_MAX_WAIT_S}s)，最后一次: {last}")
-    sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

tests/test_jimeng.py

@@ -1,153 +0,0 @@
-"""测试 ji_meng 任务工具 — 通过 Router API 调用（范本同 test_liblibai_tool.py）
-
-用法：
-    1. 先启动 Router：uv run python -m tool_agent
-    2. 运行测试：python tests/test_jimeng.py
-
-需已注册 ji_meng_add_task、ji_meng_query_task（或改下方常量）；搜索关键词默认可匹配二者。
-"""
-
-import sys
-import time
-
-if sys.platform == 'win32':
-    sys.stdout.reconfigure(encoding='utf-8')
-
-import httpx
-
-ROUTER_URL = "http://127.0.0.1:8001"
-POLL_INTERVAL_S = 2
-POLL_MAX_WAIT_S = 300
-
-
-def _extract_task_id(data: dict):
-    for key in ("task_id", "taskId", "id", "job_id", "jobId"):
-        v = data.get(key)
-        if v is not None and str(v).strip():
-            return str(v).strip()
-    inner = data.get("data")
-    if isinstance(inner, dict):
-        return _extract_task_id(inner)
-    return None
-
-
-def _terminal_success(data: dict) -> bool:
-    status = (
-        data.get("status")
-        or data.get("task_status")
-        or data.get("taskStatus")
-        or data.get("state")
-    )
-    if status is None and isinstance(data.get("data"), dict):
-        status = data["data"].get("status") or data["data"].get("task_status")
-    if status is None:
-        return False
-    s = str(status).lower()
-    return s in ("completed", "success", "done", "finished", "succeed", "complete")
-
-
-def _terminal_failure(data: dict) -> bool:
-    status = data.get("status") or data.get("task_status") or data.get("state")
-    if status is None and isinstance(data.get("data"), dict):
-        status = data["data"].get("status")
-    if status is None:
-        return False
-    s = str(status).lower()
-    return s in ("failed", "error", "cancelled", "canceled")
-
-
-def main():
-    print("=" * 50)
-    print("测试 ji_meng 任务工具")
-    print("=" * 50)
-
-    # 1. 检查 Router 是否在线
-    try:
-        resp = httpx.get(f"{ROUTER_URL}/health", timeout=3)
-        print(f"Router 状态: {resp.json()}")
-    except httpx.ConnectError:
-        print(f"无法连接 Router ({ROUTER_URL})")
-        print("请先启动: uv run python -m tool_agent")
-        sys.exit(1)
-
-    # 2. 搜索工具，确认已注册
-    print("\n--- 搜索工具 ---")
-    resp = httpx.post(f"{ROUTER_URL}/search_tools", json={"keyword": "ji_meng"})
-    tools = resp.json()
-    print(f"找到 {tools['total']} 个工具")
-    for t in tools["tools"]:
-        print(f"  {t['tool_id']}: {t['name']} (state={t['state']})")
-
-    # 3. 创建 ji_meng_add_task 工具
-    print("\n--- 调用 ji_meng_add_task 工具创建任务 ---")
-    print(f"提示词: simple white line art, cat, black background")
-    print("提交中...")
-
-    resp = httpx.post(
-        f"{ROUTER_URL}/select_tool",
-        json={
-            "tool_id": "ji_meng_add_task",
-            "params": {
-                "task_type": "image",
-                "prompt": "simple white line art, cat, black background",
-            },
-        },
-        timeout=120,
-    )
-
-    result = resp.json()
-    print(f"\n响应状态: {result.get('status')}")
-
-    if result.get("code") != 0:
-        print(f"错误: {result.get('msg')}")
-        sys.exit(1)
-
-    task_id = result.get("data", {}).get("task_id")
-    if not task_id:
-        print("错误: 无法从创建任务响应中解析 task_id")
-        sys.exit(1)
-    print(f"任务 ID: {task_id}")
-
-    # 4. 轮询查询任务
-    print("\n--- 调用 ji_meng_query_task 工具查询任务 ---")
-    deadline = time.monotonic() + POLL_MAX_WAIT_S
-    last = {}
-
-    while time.monotonic() < deadline:
-        resp = httpx.post(
-            f"{ROUTER_URL}/select_tool",
-            json={
-                "tool_id": "ji_meng_query_task",
-                "params": {"task_id": task_id},
-            },
-            timeout=60,
-        )
-        result = resp.json()
-        print(f"\n响应状态: {result.get('status')}")
-
-        if result.get("status") != "success":
-            print(f"错误: {result.get('error')}")
-            sys.exit(1)
-
-        last = result.get("result", {})
-        if not isinstance(last, dict):
-            print(f"非 dict 结果: {last}")
-            time.sleep(POLL_INTERVAL_S)
-            continue
-        if last.get("status") == "error":
-            print(f"错误: {last.get('error')}")
-            sys.exit(1)
-
-        print(f"任务状态: {last.get('status')}")
-        print(f"最终结果: {last}")
-        print("\n测试通过!")
-        return
-
-        time.sleep(POLL_INTERVAL_S)
-
-    print(f"\n等待超时 ({POLL_MAX_WAIT_S}s)，最后一次响应: {last}")
-    sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

tests/test_jimeng_ai.py

@@ -1,145 +0,0 @@
-"""测试即梦AI工具
-
-用法：
-    uv run python tests/test_jimeng_ai.py                    # Health check
-    uv run python tests/test_jimeng_ai.py --text2image       # 测试文生图
-    uv run python tests/test_jimeng_ai.py --image2video      # 测试图生视频
-    uv run python tests/test_jimeng_ai.py --query            # 测试查询任务状态
-"""
-
-import argparse
-import json
-import sys
-import httpx
-
-BASE_URL = "http://127.0.0.1:8001"
-
-
-def check_connection():
-    try:
-        httpx.get(f"{BASE_URL}/health", timeout=3)
-    except httpx.ConnectError:
-        print(f"ERROR: Cannot connect to {BASE_URL}")
-        print("Please start the service first:")
-        print("  uv run python -m tool_agent")
-        sys.exit(1)
-
-
-def test_health():
-    print("=== Health Check ===")
-    resp = httpx.get(f"{BASE_URL}/health")
-    print(f"  Status : {resp.status_code}")
-    print(f"  Body   : {json.dumps(resp.json(), ensure_ascii=False, indent=4)}")
-    assert resp.status_code == 200
-    print("  [PASS]")
-
-
-def test_text2image():
-    print("=== Test Text to Image (Seendance 2.0) ===")
-    print("  Calling jimeng_ai...")
-    try:
-        resp = httpx.post(f"{BASE_URL}/select_tool", json={
-            "tool_id": "jimeng_ai",
-            "params": {
-                "action": "text2image",
-                "prompt": "beautiful sunset over mountains, high quality",
-                "model": "seendance_2.0",
-                "aspect_ratio": "16:9",
-                "image_count": 1,
-                "cfg_scale": 7.0,
-                "steps": 20
-            }
-        }, timeout=120)
-        print(f"  Status : {resp.status_code}")
-        data = resp.json()
-        if data["status"] == "success":
-            result = data["result"]
-            print(f"  task_id: {result.get('task_id')}")
-            print(f"  status : {result.get('status')}")
-            if result.get("result", {}).get("images"):
-                print(f"  images : {len(result['result']['images'])} generated")
-            print("  [PASS]")
-        else:
-            print(f"  ERROR : {data.get('error')}")
-            print("  [FAIL]")
-    except Exception as e:
-        print(f"  ERROR : {e}")
-        print("  [FAIL]")
-
-
-def test_image2video():
-    print("=== Test Image to Video (Seedream Lite 5.0) ===")
-    print("  Calling jimeng_ai...")
-    try:
-        resp = httpx.post(f"{BASE_URL}/select_tool", json={
-            "tool_id": "jimeng_ai",
-            "params": {
-                "action": "image2video",
-                "image_url": "https://example.com/sample.jpg",
-                "prompt": "camera slowly zooming in",
-                "model": "seedream_lite_5.0",
-                "video_duration": 5,
-                "motion_strength": 0.5
-            }
-        }, timeout=300)
-        print(f"  Status : {resp.status_code}")
-        data = resp.json()
-        if data["status"] == "success":
-            result = data["result"]
-            print(f"  task_id: {result.get('task_id')}")
-            print(f"  status : {result.get('status')}")
-            if result.get("result", {}).get("videos"):
-                print(f"  videos : {len(result['result']['videos'])} generated")
-            print("  [PASS]")
-        else:
-            print(f"  ERROR : {data.get('error')}")
-            print("  [FAIL]")
-    except Exception as e:
-        print(f"  ERROR : {e}")
-        print("  [FAIL]")
-
-
-def test_query_status():
-    print("=== Test Query Task Status ===")
-    print("  NOTE: Requires a valid task_id from previous generation")
-    print("  [SKIP - Manual test required]")
-
-
-def main():
-    parser = argparse.ArgumentParser(description="Jimeng AI Tool Test")
-    parser.add_argument("--text2image", action="store_true", help="test text to image")
-    parser.add_argument("--image2video", action="store_true", help="test image to video")
-    parser.add_argument("--query", action="store_true", help="test query status")
-    args = parser.parse_args()
-
-    print(f"Target: {BASE_URL}\n")
-    check_connection()
-    test_health()
-
-    ran_any = False
-
-    if args.text2image:
-        print()
-        test_text2image()
-        ran_any = True
-
-    if args.image2video:
-        print()
-        test_image2video()
-        ran_any = True
-
-    if args.query:
-        print()
-        test_query_status()
-        ran_any = True
-
-    if not ran_any:
-        print()
-        print("No test specified. Available options:")
-        parser.print_help()
-
-    print("\n=== DONE ===")
-
-
-if __name__ == "__main__":
-    main()

tests/test_knowhub_api_integration.py

@@ -1,106 +0,0 @@
-"""测试 KnowHub 工具表 API 集成
-
-验证 FastAPI 路由是否正确调用 tool_table_tools 模块
-"""
-
-import sys
-from pathlib import Path
-
-# 添加项目根目录到 Python 路径
-project_root = Path(__file__).parent.parent
-sys.path.insert(0, str(project_root / "src"))
-
-from tool_agent.router import tool_table_tools
-
-
-def test_query_all_tools():
-    """测试查询所有工具"""
-    print("\n=== 测试：查询所有工具 ===")
-    result = tool_table_tools.query_knowhub_tools()
-
-    print(f"状态: {result['status']}")
-    print(f"工具总数: {result['total']}")
-
-    if result['total'] > 0:
-        print(f"\n前 3 个工具:")
-        for tool in result['tools'][:3]:
-            print(f"  - {tool['title']} ({tool['tool_id']})")
-            print(f"    分类: {tool['category']}, 状态: {tool['status']}")
-
-
-def test_query_by_category():
-    """测试按分类查询"""
-    print("\n=== 测试：按分类查询 (plugin) ===")
-    result = tool_table_tools.query_knowhub_tools(category="plugin")
-
-    print(f"状态: {result['status']}")
-    print(f"plugin 类工具数: {result['total']}")
-
-    for tool in result['tools'][:5]:
-        print(f"  - {tool['title']} | 状态: {tool['status']}")
-
-
-def test_query_by_keyword():
-    """测试关键词搜索"""
-    print("\n=== 测试：关键词搜索 (comfyui) ===")
-    result = tool_table_tools.query_knowhub_tools(keyword="comfyui")
-
-    print(f"状态: {result['status']}")
-    print(f"匹配工具数: {result['total']}")
-
-    for tool in result['tools']:
-        print(f"  - {tool['title']} ({tool['tool_id']})")
-        print(f"    状态: {tool['status']}")
-
-
-def test_get_tool_detail():
-    """测试获取工具详情"""
-    print("\n=== 测试：获取工具详情 (tools/image_gen/comfyui) ===")
-    result = tool_table_tools.get_knowhub_tool_detail("tools/image_gen/comfyui")
-
-    print(f"状态: {result['status']}")
-    if result['status'] == 'success':
-        tool = result['tool']
-        print(f"标题: {tool['title']}")
-        print(f"分类: {tool['category']}")
-        print(f"接入状态: {tool['status']}")
-        print(f"描述: {tool['description']}")
-        print(f"使用方法: {tool['usage']}")
-        print(f"应用场景: {tool['scenarios']}")
-        print(f"关联知识数: {len(tool['knowledge_ids'])}")
-
-
-def test_update_comfyui_status():
-    """测试更新 ComfyUI 状态为已接入"""
-    print("\n=== 测试：更新 ComfyUI 状态为已接入 ===")
-    result = tool_table_tools.update_knowhub_tool_status(
-        "tools/image_gen/comfyui", "已接入"
-    )
-
-    print(f"状态: {result['status']}")
-    print(f"消息: {result['message']}")
-
-    # 验证
-    print("\n验证更新结果:")
-    detail = tool_table_tools.get_knowhub_tool_detail("tools/image_gen/comfyui")
-    if detail['status'] == 'success':
-        print(f"当前状态: {detail['tool']['status']}")
-
-
-def main():
-    print("=" * 60)
-    print("KnowHub 工具表 API 集成测试")
-    print("=" * 60)
-
-    test_query_all_tools()
-    test_query_by_category()
-    test_query_by_keyword()
-    test_get_tool_detail()
-    test_update_comfyui_status()
-
-    print("\n" + "=" * 60)
-    print("测试完成！")
-
-
-if __name__ == "__main__":
-    main()

tests/test_knowhub_modify.py

@@ -1,85 +0,0 @@
-"""批量给 KnowHub 工具表添加 toolhub_items 字段
-
-规则：
-- 已接入的工具：填入对应的 Tool Agent 工具 ID 和名称
-- 未接入的工具：只加空的 toolhub_items 字段
-"""
-
-import httpx
-import os
-import json
-
-KNOWHUB_API = os.getenv("KNOWHUB_API", "http://43.106.118.91:9999")
-TIMEOUT = 60.0
-
-# Tool Agent registry 中已接入的工具 → KnowHub 工具表的映射
-# key: KnowHub tool_id, value: toolhub_items 列表
-TOOLHUB_MAPPING = {
-    "tools/image_gen/comfyui": [
-        {"launch_comfy_env": "启动云端 ComfyUI Docker 环境，返回 server_id 和 comfy_url"},
-        {"runcomfy_workflow_executor": "在已就绪的 RunComfy 机器上提交 ComfyUI 工作流，上传输入文件，监听执行状态，下载结果图片"},
-        {"runcomfy_stop_env": "停止并删除 RunComfy 机器实例，释放资源"},
-    ],
-}
-
-
-def get_all_tools() -> list[dict]:
-    """获取 KnowHub 所有工具"""
-    resp = httpx.get(f"{KNOWHUB_API}/api/resource", params={"limit": 1000}, timeout=TIMEOUT)
-    resp.raise_for_status()
-    data = resp.json()
-    results = data.get("results", [])
-    return [r for r in results if isinstance(r, dict) and r.get("id", "").startswith("tools/")]
-
-
-def patch_tool(tool_id: str, metadata: dict) -> bool:
-    """PATCH 更新工具 metadata"""
-    resp = httpx.patch(
-        f"{KNOWHUB_API}/api/resource/{tool_id}",
-        json={"metadata": metadata},
-        timeout=TIMEOUT
-    )
-    resp.raise_for_status()
-    return True
-
-
-def main():
-    print(f"KnowHub API: {KNOWHUB_API}")
-    print("=" * 60)
-
-    tools = get_all_tools()
-    print(f"共 {len(tools)} 个工具\n")
-
-    updated = 0
-    skipped = 0
-    failed = 0
-
-    for tool in tools:
-        tool_id = tool["id"]
-        metadata = tool.get("metadata", {})
-
-        # 已经有 toolhub_items 且非空的跳过
-        if metadata.get("toolhub_items"):
-            print(f"  SKIP  {tool_id} (已有 toolhub_items)")
-            skipped += 1
-            continue
-
-        # 查映射表，有则填入，无则空列表
-        items = TOOLHUB_MAPPING.get(tool_id, [])
-        metadata["toolhub_items"] = items
-
-        try:
-            patch_tool(tool_id, metadata)
-            tag = f"{len(items)} items" if items else "empty"
-            print(f"  OK    {tool_id} ({tag})")
-            updated += 1
-        except Exception as e:
-            print(f"  FAIL  {tool_id}: {e}")
-            failed += 1
-
-    print(f"\n{'=' * 60}")
-    print(f"完成: 更新 {updated}, 跳过 {skipped}, 失败 {failed}")
-
-
-if __name__ == "__main__":
-    main()

tests/test_knowhub_query.py

@@ -1,243 +0,0 @@
-"""测试 KnowHub 工具表查询功能"""
-
-import httpx
-import os
-from typing import Any
-
-# KnowHub API 地址
-KNOWHUB_API = os.getenv("KNOWHUB_API", "http://43.106.118.91:9999")
-
-
-def test_list_all_resources():
-    """测试：列出所有资源"""
-    print("\n=== 测试：列出所有资源 ===")
-    try:
-        resp = httpx.get(f"{KNOWHUB_API}/api/resource", params={"limit": 1000}, timeout=60.0)
-        resp.raise_for_status()
-        data = resp.json()
-
-        print(f"状态码: {resp.status_code}")
-        print(f"返回数据类型: {type(data)}")
-
-        # 检查返回格式
-        if isinstance(data, dict):
-            print(f"返回字典的键: {data.keys()}")
-            # API 返回格式: {"results": [...], "count": N}
-            if "results" in data:
-                data = data["results"]
-            elif "data" in data:
-                data = data["data"]
-
-        if not isinstance(data, list):
-            print(f"⚠️ 返回数据不是列表")
-            return []
-
-        print(f"总资源数: {len(data)}")
-
-        # 过滤出工具
-        tools = [r for r in data if isinstance(r, dict) and r.get("id", "").startswith("tools/")]
-        print(f"工具数量: {len(tools)}")
-
-        if tools:
-            print("\n前 3 个工具:")
-            for tool in tools[:3]:
-                print(f"  - {tool['id']}: {tool.get('title', 'N/A')}")
-
-        return tools
-    except Exception as e:
-        print(f"❌ 错误: {e}")
-        import traceback
-        traceback.print_exc()
-        return []
-
-
-def test_get_tool_detail(tool_id: str):
-    """测试：获取工具详情"""
-    print(f"\n=== 测试：获取工具详情 ({tool_id}) ===")
-    try:
-        resp = httpx.get(f"{KNOWHUB_API}/api/resource/{tool_id}", timeout=10.0)
-        resp.raise_for_status()
-        tool = resp.json()
-
-        print(f"状态码: {resp.status_code}")
-        print(f"ID: {tool['id']}")
-        print(f"标题: {tool.get('title', 'N/A')}")
-
-        metadata = tool.get("metadata", {})
-        print(f"\nMetadata:")
-        for key, value in metadata.items():
-            # 长列表截断显示
-            if isinstance(value, list) and len(value) > 3:
-                print(f"  - {key}: [{len(value)} items] {value[:3]}...")
-            else:
-                print(f"  - {key}: {value}")
-
-        return tool
-    except Exception as e:
-        print(f"❌ 错误: {e}")
-        return None
-
-
-def test_search_tools_by_category(category: str):
-    """测试：按分类搜索工具"""
-    print(f"\n=== 测试：按分类搜索工具 (category={category}) ===")
-    try:
-        resp = httpx.get(f"{KNOWHUB_API}/api/resource", params={"limit": 1000}, timeout=60.0)
-        resp.raise_for_status()
-        data = resp.json()
-
-        # 检查返回格式
-        if isinstance(data, dict):
-            # API 返回格式: {"results": [...], "count": N}
-            if "results" in data:
-                data = data["results"]
-            elif "data" in data:
-                data = data["data"]
-
-        if not isinstance(data, list):
-            print(f"⚠️ 返回数据不是列表")
-            return []
-
-        # 客户端过滤
-        tools = [
-            r for r in data
-            if isinstance(r, dict) and r.get("id", "").startswith("tools/")
-            and r.get("metadata", {}).get("category") == category
-        ]
-
-        print(f"找到 {len(tools)} 个 {category} 类工具:")
-        for tool in tools[:5]:
-            print(f"  - {tool['id']}: {tool.get('title', 'N/A')}")
-
-        return tools
-    except Exception as e:
-        print(f"❌ 错误: {e}")
-        import traceback
-        traceback.print_exc()
-        return []
-
-
-def test_search_knowledge_by_tool(query: str, top_k: int = 5):
-    """测试：搜索工具相关知识"""
-    print(f"\n=== 测试：搜索工具相关知识 (query={query}) ===")
-    try:
-        resp = httpx.get(
-            f"{KNOWHUB_API}/api/knowledge/search",
-            params={"q": query, "top_k": top_k, "min_score": 3},
-            timeout=60.0
-        )
-        resp.raise_for_status()
-        results = resp.json()
-
-        print(f"状态码: {resp.status_code}")
-        print(f"返回数据类型: {type(results)}")
-
-        # 检查返回格式
-        if isinstance(results, dict):
-            print(f"返回字典的键: {results.keys()}")
-            if "results" in results:
-                results = results["results"]
-            elif "data" in results:
-                results = results["data"]
-
-        if not isinstance(results, list):
-            print(f"⚠️ 返回数据不是列表")
-            return []
-
-        print(f"找到 {len(results)} 条知识:")
-
-        for i, item in enumerate(results[:3], 1):
-            if isinstance(item, dict):
-                print(f"\n  {i}. {item.get('id', 'N/A')}")
-                title = item.get('title', 'N/A')
-                print(f"     标题: {title[:50] if isinstance(title, str) else title}")
-                print(f"     分数: {item.get('score', 0):.2f}")
-                print(f"     关联工具: {item.get('resource_ids', [])}")
-
-        return results
-    except Exception as e:
-        print(f"❌ 错误: {e}")
-        import traceback
-        traceback.print_exc()
-        return []
-
-
-def test_list_tool_knowledge(tags: str = "tool"):
-    """测试：列出所有工具相关知识"""
-    print(f"\n=== 测试：列出所有工具相关知识 (tags={tags}) ===")
-    try:
-        resp = httpx.get(
-            f"{KNOWHUB_API}/api/knowledge",
-            params={"tags": tags, "status": "approved,checked", "page_size": 200},
-            timeout=60.0
-        )
-        resp.raise_for_status()
-        results = resp.json()
-
-        print(f"状态码: {resp.status_code}")
-        print(f"返回数据类型: {type(results)}")
-
-        # 检查返回格式
-        if isinstance(results, dict):
-            print(f"返回字典的键: {results.keys()}")
-            if "results" in results:
-                results = results["results"]
-            elif "data" in results:
-                results = results["data"]
-            elif "items" in results:
-                results = results["items"]
-
-        if not isinstance(results, list):
-            print(f"⚠️ 返回数据不是列表")
-            return []
-
-        print(f"找到 {len(results)} 条工具知识")
-
-        if results:
-            print("\n前 3 条:")
-            for item in results[:3]:
-                if isinstance(item, dict):
-                    title = item.get('title', 'N/A')
-                    print(f"  - {item.get('id', 'N/A')}: {title[:40] if isinstance(title, str) else title}")
-                    print(f"    关联工具: {item.get('resource_ids', [])}")
-
-        return results
-    except Exception as e:
-        print(f"❌ 错误: {e}")
-        import traceback
-        traceback.print_exc()
-        return []
-
-
-def main():
-    """运行所有测试"""
-    print(f"KnowHub API: {KNOWHUB_API}")
-    print("=" * 60)
-
-    # 1. 列出所有资源
-    tools = test_list_all_resources()
-
-    # 2. 获取 ComfyUI 工具详情（验证 toolhub_items）
-    test_get_tool_detail("tools/image_gen/comfyui")
-
-    # 3. 获取第一个工具的详情
-    if tools:
-        first_tool_id = tools[0]["id"]
-        if first_tool_id != "tools/image_gen/comfyui":
-            test_get_tool_detail(first_tool_id)
-
-    # 3. 按分类搜索
-    test_search_tools_by_category("plugin")
-
-    # 4. 搜索工具相关知识
-    test_search_knowledge_by_tool("IP-Adapter")
-
-    # 5. 列出所有工具知识
-    test_list_tool_knowledge()
-
-    print("\n" + "=" * 60)
-    print("测试完成！")
-
-
-if __name__ == "__main__":
-    main()

tests/test_kuaishou_kling.py

@@ -1,146 +0,0 @@
-"""测试快手可灵AI工具
-
-用法：
-    uv run python tests/test_kuaishou_kling.py                    # Health check
-    uv run python tests/test_kuaishou_kling.py --text2image       # 测试文生图
-    uv run python tests/test_kuaishou_kling.py --text2video       # 测试文生视频
-    uv run python tests/test_kuaishou_kling.py --lip-sync         # 测试对口型
-"""
-
-import argparse
-import json
-import sys
-import httpx
-
-BASE_URL = "http://127.0.0.1:8001"
-
-
-def check_connection():
-    try:
-        httpx.get(f"{BASE_URL}/health", timeout=3)
-    except httpx.ConnectError:
-        print(f"ERROR: Cannot connect to {BASE_URL}")
-        print("Please start the service first:")
-        print("  uv run python -m tool_agent")
-        sys.exit(1)
-
-
-def test_health():
-    print("=== Health Check ===")
-    resp = httpx.get(f"{BASE_URL}/health")
-    print(f"  Status : {resp.status_code}")
-    print(f"  Body   : {json.dumps(resp.json(), ensure_ascii=False, indent=4)}")
-    assert resp.status_code == 200
-    print("  [PASS]")
-
-
-def test_text2image():
-    print("=== Test Text to Image ===")
-    print("  Calling kuaishou_kling...")
-    try:
-        resp = httpx.post(f"{BASE_URL}/select_tool", json={
-            "tool_id": "kuaishou_kling",
-            "params": {
-                "biz_type": "aiImage",
-                "prompt": "cute cat playing in garden",
-                "aspect_ratio": "16:9",
-                "image_count": 2
-            }
-        }, timeout=120)
-        print(f"  Status : {resp.status_code}")
-        data = resp.json()
-        if data["status"] == "success":
-            result = data["result"]
-            print(f"  task_id: {result.get('task_id')}")
-            print("  [PASS]")
-        else:
-            print(f"  ERROR : {data.get('error')}")
-            print("  [FAIL]")
-    except Exception as e:
-        print(f"  ERROR : {e}")
-        print("  [FAIL]")
-
-
-def test_text2video():
-    print("=== Test Text to Video ===")
-    print("  Calling kuaishou_kling...")
-    try:
-        resp = httpx.post(f"{BASE_URL}/select_tool", json={
-            "tool_id": "kuaishou_kling",
-            "params": {
-                "biz_type": "aiVideo",
-                "prompt": "ocean waves at sunset",
-                "aspect_ratio": "16:9"
-            }
-        }, timeout=300)
-        print(f"  Status : {resp.status_code}")
-        data = resp.json()
-        if data["status"] == "success":
-            print(f"  task_id: {data['result'].get('task_id')}")
-            print("  [PASS]")
-        else:
-            print(f"  ERROR : {data.get('error')}")
-            print("  [FAIL]")
-    except Exception as e:
-        print(f"  ERROR : {e}")
-        print("  [FAIL]")
-
-
-def test_lip_sync():
-    print("=== Test AI Lip Sync ===")
-    print("  Calling kuaishou_kling...")
-    try:
-        resp = httpx.post(f"{BASE_URL}/select_tool", json={
-            "tool_id": "kuaishou_kling",
-            "params": {
-                "biz_type": "aiLipSync",
-                "mode": "text2video",
-                "text": "Hello world"
-            }
-        }, timeout=300)
-        print(f"  Status : {resp.status_code}")
-        data = resp.json()
-        if data["status"] == "success":
-            print("  [PASS]")
-        else:
-            print(f"  ERROR : {data.get('error')}")
-            print("  [FAIL]")
-    except Exception as e:
-        print(f"  ERROR : {e}")
-        print("  [FAIL]")
-
-
-def main():
-    parser = argparse.ArgumentParser(description="Kuaishou Kling AI Tool Test")
-    parser.add_argument("--text2image", action="store_true")
-    parser.add_argument("--text2video", action="store_true")
-    parser.add_argument("--lip-sync", action="store_true")
-    args = parser.parse_args()
-
-    print(f"Target: {BASE_URL}\n")
-    check_connection()
-    test_health()
-
-    ran_any = False
-    if args.text2image:
-        print()
-        test_text2image()
-        ran_any = True
-    if args.text2video:
-        print()
-        test_text2video()
-        ran_any = True
-    if args.lip_sync:
-        print()
-        test_lip_sync()
-        ran_any = True
-
-    if not ran_any:
-        print("\nNo test specified. Available options:")
-        parser.print_help()
-
-    print("\n=== DONE ===")
-
-
-if __name__ == "__main__":
-    main()

tests/test_lib.py

@@ -1,158 +0,0 @@
-import requests
-import time
-import hmac
-import hashlib
-import uuid
-import base64
-import os
-import sys
-from dotenv import load_dotenv
-
-# 设置 UTF-8 输出编码（Windows 兼容）
-if sys.platform == 'win32':
-    sys.stdout.reconfigure(encoding='utf-8')
-
-load_dotenv()
-
-class LibLibControlNet:
-    def __init__(self):
-        self.ak = os.getenv("LIBLIBAI_ACCESS_KEY") or "你的AccessKey"
-        self.sk = os.getenv("LIBLIBAI_SECRET_KEY") or "你的SecretKey"
-        self.domain = "https://openapi.liblibai.cloud"
-        # 先替换为更通用的模板ID（避免旧ID失效）
-        # 文生图模板（必须是 1.5 体系的模板）
-        self.TEMPLATE_UUID = "e10adc3949ba59abbe56e057f20f883e"
-
-        # SD1.5 底模
-        self.CHECKPOINT_ID = "0ea388c7eb854be3ba3c6f65aac6bfd3"
-
-        # SD1.5 Canny（来自你给的官方表）
-        self.CANNY_MODEL_ID = "b6806516962f4e1599a93ac4483c3d23"
-
-    def _generate_signature(self, uri):
-        ts = str(int(time.time() * 1000))
-        nonce = uuid.uuid4().hex
-        sign_str = f"{uri}&{ts}&{nonce}"
-        dig = hmac.new(self.sk.encode(), sign_str.encode(), hashlib.sha1).digest()
-        signature = base64.urlsafe_b64encode(dig).rstrip(b"=").decode()
-        return f"{self.domain}{uri}?AccessKey={self.ak}&Timestamp={ts}&SignatureNonce={nonce}&Signature={signature}"
-
-    def submit_canny_task(self, image_url, prompt):
-        uri = "/api/generate/webui/text2img"
-        url = self._generate_signature(uri)
-        
-        payload = {
-            "templateUuid": self.TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": self.CHECKPOINT_ID,
-                "prompt": prompt,
-                "negativePrompt": "lowres, bad anatomy, text, error, extra digit, fewer digits, cropped, worst quality, low quality",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,  # 修复：改为浮点数（部分版本要求）
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "seed": -1,
-                "controlNet": [
-                    {
-                        "unitOrder": 1,
-                        "sourceImage": image_url,
-                        "width": 512,
-                        "height": 512,
-                        "preprocessor": 1,
-                        "annotationParameters": {
-                            "canny": {
-                                "preprocessorResolution": 512,
-                                "lowThreshold": 100,
-                                "highThreshold": 200
-                            }
-                        },
-                        "model": self.CANNY_MODEL_ID,
-                        "controlWeight": 1.0,
-                        "startingControlStep": 0.0,
-                        "endingControlStep": 1.0,
-                        "pixelPerfect": 1,
-                        "controlMode": 0,
-                        "resizeMode": 1
-                    }
-                ]
-            }
-        }
-
-        try:
-            print(f"📤 提交ControlNet生图任务...")
-            print(f"🔍 请求URL: {url}")
-            print(f"🔍 请求Payload: {payload}")  # 打印完整参数
-            resp = requests.post(
-                url,
-                headers={"Content-Type": "application/json"},
-                json=payload,
-                timeout=10
-            )
-            # 打印完整响应（关键调试信息）
-            print(f"🔍 响应状态码: {resp.status_code}")
-            print(f"🔍 响应内容: {resp.text}")
-            
-            result = resp.json()
-            if result.get("code") == 0:
-                task_id = result["data"]["generateUuid"]
-                print(f"✅ 任务提交成功！任务ID: {task_id}")
-                return task_id
-            else:
-                print(f"❌ 任务提交失败: {result.get('msg')}")
-                return None
-        except Exception as e:
-            print(f"❌ 提交任务异常: {str(e)}")
-            return None
-
-    def poll_task_status(self, task_id):
-        uri = "/api/generate/webui/status"
-        max_retry = 60
-        retry_count = 0
-
-        while retry_count < max_retry:
-            try:
-                url = self._generate_signature(uri)
-                resp = requests.post(
-                    url,
-                    headers={"Content-Type": "application/json"},
-                    json={"generateUuid": task_id},
-                    timeout=10
-                )
-                resp.raise_for_status()
-                result = resp.json()
-
-                if result.get("code") != 0:
-                    print(f"❌ 查询状态失败: {result.get('msg')}")
-                    return None
-
-                data = result.get("data", {})
-                status = data.get("generateStatus")
-                if status == 5:
-                    img_url = data["images"][0]["imageUrl"]
-                    print(f"🎉 生图成功！最终图片URL: {img_url}")
-                    return img_url
-                elif status in [6, 7]:
-                    print(f"❌ 任务失败: {data.get('generateMsg', '未知错误')}")
-                    return None
-                else:
-                    print(f"⏳ 任务处理中（状态码: {status}），已等待{retry_count*5}秒...")
-            except Exception as e:
-                print(f"❌ 轮询异常: {str(e)}")
-            
-            retry_count += 1
-            time.sleep(5)
-        
-        print(f"❌ 任务轮询超时（超过5分钟）")
-        return None
-
-if __name__ == "__main__":
-    client = LibLibControlNet()
-    # 替换为更稳定的公开测试图
-    PUBLIC_IMAGE_URL = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-    PROMPT = "simple white line art, cat, black background, 512x512"  # 简化提示词（避免过长）
-    
-    task_id = client.submit_canny_task(PUBLIC_IMAGE_URL, PROMPT)
-    if task_id:
-        client.poll_task_status(task_id)

tests/test_liblibai_tool.py

@@ -1,91 +0,0 @@
-"""测试 liblibai_controlnet 工具 — 通过 Router API 调用
-
-用法：
-    1. 先启动 Router：uv run python -m tool_agent
-    2. 运行测试：python tests/test_liblibai_tool.py
-"""
-
-import sys
-import time
-
-if sys.platform == 'win32':
-    sys.stdout.reconfigure(encoding='utf-8')
-
-import httpx
-
-ROUTER_URL = "http://127.0.0.1:8001"
-IMAGE_URL = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-
-
-def main():
-    print("=" * 50)
-    print("测试 liblibai_controlnet 工具")
-    print("=" * 50)
-
-    # 1. 检查 Router 是否在线
-    try:
-        resp = httpx.get(f"{ROUTER_URL}/health", timeout=3)
-        print(f"Router 状态: {resp.json()}")
-    except httpx.ConnectError:
-        print(f"无法连接 Router ({ROUTER_URL})")
-        print("请先启动: uv run python -m tool_agent")
-        sys.exit(1)
-
-    # 2. 搜索工具，确认已注册
-    print("\n--- 获取工具 ---")
-    resp = httpx.get(f"{ROUTER_URL}/tools")
-    tools = resp.json()
-    liblib_tools = [t for t in tools["tools"] if "liblib" in t["tool_id"] or "liblib" in t["name"].lower()]
-    print(f"找到 {len(liblib_tools)} 个匹配 liblib 的工具")
-    for t in liblib_tools:
-        print(f"  {t['tool_id']}: {t['name']} (state={t['state']})")
-
-    # 3. 调用 liblibai_controlnet 工具
-    print("\n--- 调用 ControlNet 生图 ---")
-    print(f"图片: {IMAGE_URL}")
-    print(f"提示词: simple white line art, cat, black background")
-    print("提交中...")
-
-    resp = httpx.post(
-        f"{ROUTER_URL}/run_tool",
-        json={
-            "tool_id": "liblibai_controlnet",
-            "params": {
-                "image": IMAGE_URL,
-                "prompt": "simple white line art, cat, black background",
-                "negative_prompt": "lowres, bad anatomy, text, error",
-                "width": 512,
-                "height": 512,
-                "steps": 20,
-                "cfg_scale": 7,
-                "img_count": 1,
-                "control_weight": 1.0,
-                "preprocessor": 1,
-                "canny_low": 100,
-                "canny_high": 200
-            }
-        },
-        timeout=300  # 生图可能需要几分钟
-    )
-
-    result = resp.json()
-    print(f"\n响应状态: {result.get('status')}")
-
-    if result.get("status") == "success":
-        data = result.get("result", {})
-        print(f"任务状态: {data.get('status')}")
-        print(f"任务 ID: {data.get('task_id')}")
-        images = data.get("images", [])
-        if images:
-            print(f"生成图片数: {len(images)}")
-            for i, url in enumerate(images):
-                print(f"  [{i+1}] {url}")
-            print("\n测试通过!")
-        else:
-            print("未返回图片，可能任务仍在处理中")
-    else:
-        print(f"错误: {result.get('error')}")
-
-
-if __name__ == "__main__":
-    main()

tests/test_liblibai_workflows.py

@@ -1,466 +0,0 @@
-import os
-import sys
-import time
-import requests
-import json
-from typing import Dict, Any
-
-sys.path.append(os.path.join(os.path.dirname(__file__), '../tools/local/liblibai_controlnet'))
-from liblibai_client import LibLibAIClient
-
-# 常量配置 (根据文档)
-TEMPLATE_UUID = "e10adc3949ba59abbe56e057f20f883e"  # SD1.5 & SDXL 通用自定义参数模板
-
-# 请确保这是一个有效的 SDXL 模型，这样才能匹配底下的 SDXL Canny 模型
-# 这里用的是代码里原本的 Checkpoint ID，请根据你们自己的 Liblib 模型库调整！
-DEFAULT_CHECKPOINT_ID = "0ea388c7eb854be3ba3c6f65aac6bfd3"
-
-class LibLibTestRunner:
-    def __init__(self):
-        self.client = LibLibAIClient()
-        self.models = self._load_models_from_json()
-        
-        # 动态匹配基础算法 XL 的各种控制网模型
-        self.sdxl_canny = self._get_model_uuid("线稿类", "Canny（硬边缘）", xl_only=True) or "b6806516962f4e1599a93ac4483c3d23"
-        self.sdxl_softedge = self._get_model_uuid("线稿类", "SoftEdge（软边缘）", xl_only=True) or "dda1a0c480bfab9833d9d9a1e4a71fff"
-        self.sdxl_lineart = self._get_model_uuid("线稿类", "Lineart（线稿）", xl_only=True) or "a0f01da42bf48b0ba02c86b6c26b5699"
-        self.sdxl_openpose = self._get_model_uuid("姿态类", "OpenPose（姿态）", xl_only=True) or "2fe4f992a81c5ccbdf8e9851c8c96ff2"
-        
-        self._verify_models()
-        
-    def _load_models_from_json(self):
-        json_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../data/liblibai_controlnet_models.json"))
-        if os.path.exists(json_path):
-            with open(json_path, "r", encoding="utf-8") as f:
-                return json.load(f)
-        return {}
-
-    def _get_model_uuid(self, category, subtype, xl_only=True):
-        if category in self.models and subtype in self.models[category]:
-            for model in self.models[category][subtype]:
-                if xl_only and model["base_algorithm"] != "基础算法 XL":
-                    continue
-                return model["uuid"]
-        return None
-
-    def _verify_models(self):
-        print("="*50)
-        print("校验底模信息 (使用 api/model/version/get)")
-        print("="*50)
-        
-        models_to_verify = {
-            "CheckPoint / 底模": DEFAULT_CHECKPOINT_ID,
-        }
-        
-        for name, uuid in models_to_verify.items():
-            info = self.client.get_model_version_info(uuid)
-            status = f"{info.get('model_name', 'Unknown')} (Base: {info.get('baseAlgo', 'Unknown')})" if info else "Failed to verify"
-            print(f"- {name}: [{uuid}] -> {status}")
-        print("\n")
-        
-    def _submit_task(self, payload: dict) -> str:
-        url = self.client.generate_auth_url("/api/generate/webui/text2img")
-        print(f"Submitting payload...")
-        resp = requests.post(url, json=payload, timeout=10)
-        data = resp.json()
-        if data.get("code") != 0:
-            raise Exception(f"Submit task failed: {data.get('msg')} (code: {data.get('code')})")
-        return data["data"]["generateUuid"]
-
-    def _wait_and_print_result(self, task_id: str):
-        print(f"Task submitted successfully! Task ID: {task_id}")
-        print("Waiting for result...")
-        timeout = 300
-        start_time = time.time()
-        while time.time() - start_time < timeout:
-            task_data = self.client.query_task_status(task_id)
-            status = task_data.get("generateStatus")
-            
-            if status == 5: # Success
-                images = [img["imageUrl"] for img in task_data.get("images", [])]
-                print(f"\n[SUCCESSS] Generated images: {images}")
-                
-                # --- 新增: 自动下载图片 ---
-                output_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "output"))
-                os.makedirs(output_dir, exist_ok=True)
-                for i, img_url in enumerate(images):
-                    try:
-                        img_resp = requests.get(img_url)
-                        img_resp.raise_for_status()
-                        timestamp = int(time.time())
-                        filename = f"workflow_result_{timestamp}_{i}.png"
-                        filepath = os.path.join(output_dir, filename)
-                        with open(filepath, "wb") as f:
-                            f.write(img_resp.content)
-                        print(f"[{i}] 已自动保存到本地: {filepath}")
-                    except Exception as e:
-                        print(f"图片自动下载失败: {e}")
-                return
-            elif status in [6, 7]: # Failed or Cancelled
-                print(f"\n[FAILED] Task failed. Audit status: {task_data.get('auditStatus')}")
-                return
-                
-            print(".", end="", flush=True)
-            time.sleep(5)
-            
-        print("\n[TIMEOUT] Wait for task timed out.")
-
-    # 1. 纯文生图测试
-    def test_text2img(self):
-        print("\n" + "="*50)
-        print("TEST 1: 纯文生图 (Text to Image)")
-        print("="*50)
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "prompt": "1girl, cute, beautiful landscape, masterpiece",
-                "negativePrompt": "lowres, bad anatomy, text, error",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 2. 纯图生图测试
-    def test_img2img(self):
-        print("\n" + "="*50)
-        print("TEST 2: 纯图生图 (Image to Image)")
-        print("="*50)
-        
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "mode": 0,  # 0 代表图生图
-                "sourceImage": test_img_url,
-                "denoisingStrength": 0.75, # 去噪强度
-                "prompt": "white line art, cat",
-                "negativePrompt": "lowres, bad anatomy, error",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 3. 文生图 + ControlNet (Canny) 测试
-    def test_text2img_controlnet(self):
-        print("\n" + "="*50)
-        print("TEST 3: 文生图 + ControlNet (Text2Img with ControlNet Canny)")
-        print("="*50)
-        print("注意: 如果默认 Checkpoint 是 SD1.5, 下面配置的 SDXL Canny 模型可能会导致访问拒绝!")
-        
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "prompt": "simple white line art, cat, black background",
-                "negativePrompt": "lowres, bad anatomy",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "controlNet": [{
-                    "unitOrder": 1,
-                    "sourceImage": test_img_url,
-                    "width": 512,
-                    "height": 512,
-                    "preprocessor": 1,         # 1 = Canny
-                    "model": self.sdxl_canny,
-                    "controlWeight": 1.0,
-                    "startingControlStep": 0.0,
-                    "endingControlStep": 1.0,
-                    "pixelPerfect": 1,
-                    "controlMode": 0,
-                    "annotationParameters": {
-                        "canny": {
-                            "preprocessorResolution": 512,
-                            "lowThreshold": 100,
-                            "highThreshold": 200
-                        }
-                    }
-                }]
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 4. 文生图 + 边缘 (SoftEdge/HED)
-    def test_text2img_softedge(self):
-        print("\n" + "="*50)
-        print("TEST 4: 文生图 + SoftEdge (软边缘 控制网)")
-        print("="*50)
-        
-        # 测试用图片
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "prompt": "Soft edge artwork, beautiful soft lighting, cat",
-                "negativePrompt": "lowres, bad anatomy",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "controlNet": [{
-                    "unitOrder": 1,
-                    "sourceImage": test_img_url,
-                    "width": 512,
-                    "height": 512,
-                    "preprocessor": 5,         # 5 = HED / 软边缘
-                    "model": self.sdxl_softedge,
-                    "controlWeight": 1.0,
-                    "startingControlStep": 0.0,
-                    "endingControlStep": 1.0,
-                    "pixelPerfect": 1,
-                    "controlMode": 0,
-                    "annotationParameters": {
-                        "hed": {
-                            "preprocessorResolution": 512
-                        }
-                    }
-                }]
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 5. 文生图 + 线稿 (Lineart)
-    def test_text2img_lineart(self):
-        print("\n" + "="*50)
-        print("TEST 5: 文生图 + Lineart (线稿 控制网)")
-        print("="*50)
-        
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "prompt": "Detailed coloring, colorful fantasy style, masterpiece, cat",
-                "negativePrompt": "lowres",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "controlNet": [{
-                    "unitOrder": 1,
-                    "sourceImage": test_img_url,
-                    "width": 512,
-                    "height": 512,
-                    "preprocessor": 32,        # 32 = Lineart Standard
-                    "model": self.sdxl_lineart,
-                    "controlWeight": 1.0,
-                    "startingControlStep": 0.0,
-                    "endingControlStep": 1.0,
-                    "pixelPerfect": 1,
-                    "controlMode": 0,
-                    "annotationParameters": {
-                        "lineart": {
-                            "preprocessorResolution": 512
-                        }
-                    }
-                }]
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 6. 文生图 + 骨骼 (OpenPose)
-    def test_text2img_openpose(self):
-        print("\n" + "="*50)
-        print("TEST 6: 文生图 + OpenPose (骨骼 控制网)")
-        print("="*50)
-        
-        # 测试用图片（最好使用含有人物动作的图）
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        
-        payload = {
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "prompt": "1girl, dancing pose, beautiful dress",
-                "negativePrompt": "lowres",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "controlNet": [{
-                    "unitOrder": 1,
-                    "sourceImage": test_img_url,
-                    "width": 512,
-                    "height": 512,
-                    "preprocessor": 14,        # 14 = OpenPose Full
-                    "model": self.sdxl_openpose,
-                    "controlWeight": 1.0,
-                    "startingControlStep": 0.0,
-                    "endingControlStep": 1.0,
-                    "pixelPerfect": 1,
-                    "controlMode": 0,
-                    "annotationParameters": {
-                        "openposeFull": {
-                            "preprocessorResolution": 512
-                        }
-                    }
-                }]
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-    # 7. 局部重绘 (Inpaint Mode 4)
-    def test_inpaint_mode4(self):
-        print("\n" + "="*50)
-        print("TEST 7: 局部重绘 (Inpaint Mode 4 蒙版重绘)")
-        print("="*50)
-        
-        test_img_url = "https://liblibai-airship-temp.oss-cn-beijing.aliyuncs.com/aliyun-cn-prod/73ed6ae42b144d21bf566e05b5a6c138.png"
-        test_mask_url = test_img_url # 仅作演示，实际应用中应当是一个黑底白色的蒙版图片
-        
-        payload = {
-            # Inpainting 可能需要特殊模板，如不兼容请参考文档
-            "templateUuid": TEMPLATE_UUID,
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID,
-                "mode": 4,  # 4 = Inpaint 蒙版重绘
-                "sourceImage": test_img_url,
-                "denoisingStrength": 0.5,
-                "prompt": "A completely different background, sci-fi city",
-                "negativePrompt": "lowres",
-                "sampler": 15,
-                "steps": 20,
-                "cfgScale": 7.0,
-                "width": 512,
-                "height": 512,
-                "imgCount": 1,
-                "inpaintParam": {
-                    "maskImage": test_mask_url,
-                    "maskBlur": 4,
-                    "inpaintArea": 0
-                }
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-    # 8. 人像换脸 (InstantID)
-    def test_instantid_faceswap(self):
-        print("\n" + "="*50)
-        print("TEST 8: 人像换脸 (InstantID)")
-        print("="*50)
-        
-        # Note: InstantID faceswap uses a UNIQUE templateUuid
-        INSTANT_ID_TEMPLATE_UUID = "7d888009f81d4252a7c458c874cd017f"
-        
-        face_img_url = "https://liblibai-online.liblib.cloud/img/081e9f07d9bd4c2ba090efde163518f9/49943c0b-4d79-4e2f-8c55-bc1e5b8c69d8.png"
-        pose_img_url = "https://liblibai-online.liblib.cloud/img/081e9f07d9bd4c2ba090efde163518f9/e713676d-baaa-4dac-99b9-d5d814a29f9f.png"
-        
-        payload = {
-            "templateUuid": INSTANT_ID_TEMPLATE_UUID, 
-            "generateParams": {
-                "checkPointId": DEFAULT_CHECKPOINT_ID, # 仅 XL模型支持人像换脸
-                "prompt": "Asian portrait,A young woman wearing a green baseball cap, close shot, background is coffee store, masterpiece, best quality, ultra resolution",
-                "width": 768,
-                "height": 1152,
-                "sampler": 20,
-                "steps": 35,
-                "cfgScale": 2.0,
-                "imgCount": 1,
-                "controlNet": [
-                    {
-                        "unitOrder": 1, # 第一步：先识别要用的人像人脸
-                        "sourceImage": face_img_url,
-                        "width": 1080,
-                        "height": 1432            
-                    },
-                    {
-                        "unitOrder": 2, # 第二步：再识别要参考的人物面部朝向
-                        "sourceImage": pose_img_url,
-                        "width": 1024,
-                        "height": 1024 
-                    }
-                ]
-            }
-        }
-        task_id = self._submit_task(payload)
-        self._wait_and_print_result(task_id)
-
-
-if __name__ == "__main__":
-    try:
-        runner = LibLibTestRunner()
-        
-        print("选择要测试的模式:")
-        print("1. 纯文生图 (Text2Img)")
-        print("2. 纯图生图 (Img2Img)")
-        print("3. 硬边缘控制 (Canny ControlNet)")
-        print("4. 软边缘控制 (SoftEdge ControlNet)")
-        print("5. 线稿控制 (Lineart ControlNet)")
-        print("6. 骨骼控制 (OpenPose ControlNet)")
-        print("7. 局部重绘 (Inpaint Mode=4)")
-        print("8. 人像换脸 (InstantID)")
-        print("9. 全部测试 (All)")
-        
-        if len(sys.argv) > 1:
-            choice = sys.argv[1]
-        else:
-            # 默认测试文生图以验证 API Key 最基本权限
-            choice = "1"
-            
-        if choice == "1":
-            runner.test_text2img()
-        elif choice == "2":
-            runner.test_img2img()
-        elif choice == "3":
-            runner.test_text2img_controlnet()
-        elif choice == "4":
-            runner.test_text2img_softedge()
-        elif choice == "5":
-            runner.test_text2img_lineart()
-        elif choice == "6":
-            runner.test_text2img_openpose()
-        elif choice == "7":
-            runner.test_inpaint_mode4()
-        elif choice == "8":
-            runner.test_instantid_faceswap()
-        elif choice == "9":
-            runner.test_text2img()
-            runner.test_img2img()
-            runner.test_text2img_controlnet()
-            # 省略后面的测试避免同时发生太多请求...
-        else:
-            print("Unknown choice.")
-            
-    except Exception as e:
-        print(f"\n[FATAL ERROR] {e}")

tests/test_local_openpose.py

@@ -1,126 +0,0 @@
-import os
-import sys
-import time
-import base64
-import requests
-
-sys.path.append(os.path.join(os.path.dirname(__file__), '../tools/local/liblibai_controlnet'))
-from liblibai_client import LibLibAIClient
-
-TEMPLATE_UUID = "e10adc3949ba59abbe56e057f20f883e"
-DEFAULT_CHECKPOINT_ID = "0ea388c7eb854be3ba3c6f65aac6bfd3" 
-
-# 这里填入您实际能用的 OpenPose SDXL 模型的 UUID
-OPENPOSE_MODEL_ID = "b6806516962f4e1599a93ac4483c3d23" # 注意：这里默认放的是Canny，请记得更换成 OpenPose UUID
-
-
-def get_base64_image(image_path):
-    with open(image_path, "rb") as f:
-        return "data:image/png;base64," + base64.b64encode(f.read()).decode("utf-8")
-
-def main():
-    print("=" * 50)
-    print("测试: 读取本地图片并自动上传 -> 使用 OpenPose 生成图片")
-    print("=" * 50)
-
-    client = LibLibAIClient()
-    
-    # 1. 读取本地图片并转换为 Base64
-    local_image_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "features/pose_skeleton/img_1_openpose.png"))
-    print(f"Reading local image: {local_image_path}")
-    
-    if not os.path.exists(local_image_path):
-        print(f"Error: 找不到图片 {local_image_path}")
-        sys.exit(1)
-        
-    base64_image = get_base64_image(local_image_path)
-    
-    # 2. 借助已经写好的 `upload_base64_image` 上传到 OSS
-    print("Uploading Base64 to Liblib OSS...")
-    image_url = client.upload_base64_image(base64_image)
-    print(f"Uploaded successfully! Public URL: {image_url}")
-    
-    # 3. 发送任务
-    payload = {
-        "templateUuid": TEMPLATE_UUID,
-        "generateParams": {
-            "checkPointId": DEFAULT_CHECKPOINT_ID,
-            "prompt": "1girl, highly detailed, masterpieces, beautiful face, standing pose",
-            "negativePrompt": "lowres, bad anatomy",
-            "sampler": 15,
-            "steps": 20,
-            "cfgScale": 7.0,
-            "width": 512,
-            "height": 512,
-            "imgCount": 1,
-            "controlNet": [{
-                "unitOrder": 1,
-                "sourceImage": image_url,
-                "width": 512,
-                "height": 512,
-                "preprocessor": 14,        # 14 = OpenPose Full
-                "model": OPENPOSE_MODEL_ID, # !!需更换为OpenPose模型UUID
-                "controlWeight": 1.0,
-                "startingControlStep": 0.0,
-                "endingControlStep": 1.0,
-                "pixelPerfect": 1,
-                "controlMode": 0,
-                "annotationParameters": {
-                    "openposeFull": {
-                        "preprocessorResolution": 512
-                    }
-                }
-            }]
-        }
-    }
-    
-    auth_url = client.generate_auth_url("/api/generate/webui/text2img")
-    print(f"Submitting OpenPose payload...")
-    resp = requests.post(auth_url, json=payload, timeout=10)
-    data = resp.json()
-    if data.get("code") != 0:
-        print(f"Submit task failed: {data.get('msg')} (code: {data.get('code')})")
-        sys.exit(1)
-        
-    task_id = data["data"]["generateUuid"]
-    print(f"Task submitted! Task ID: {task_id}")
-    
-    timeout = 300
-    start_time = time.time()
-    while time.time() - start_time < timeout:
-        task_data = client.query_task_status(task_id)
-        status = task_data.get("generateStatus")
-        
-        if status == 5:
-            images = [img["imageUrl"] for img in task_data.get("images", [])]
-            print(f"\n[SUCCESSS] Generated images: {images}")
-            
-            # --- 新增: 自动下载图片到 output 文件夹 ---
-            output_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "output"))
-            os.makedirs(output_dir, exist_ok=True)
-            for i, img_url in enumerate(images):
-                try:
-                    img_resp = requests.get(img_url)
-                    img_resp.raise_for_status()
-                    # 以时间戳作为文件名前缀防止覆盖
-                    timestamp = int(time.time())
-                    filename = f"openpose_result_{timestamp}_{i}.png"
-                    filepath = os.path.join(output_dir, filename)
-                    with open(filepath, "wb") as f:
-                        f.write(img_resp.content)
-                    print(f"[{i}] 已自动下载到本地: {filepath}")
-                except Exception as e:
-                    print(f"[{i}] 图片自动下载失败: {e}")
-                    
-            return
-        elif status in [6, 7]:
-            print("\n[FAILED] Task failed.")
-            return
-            
-        print(".", end="", flush=True)
-        time.sleep(5)
-        
-    print("\n[TIMEOUT]")
-
-if __name__ == "__main__":
-    main()

tests/test_local_runner.py

@@ -1,10 +0,0 @@
-"""测试 LocalRunner"""
-
-import pytest
-from tool_agent.runtime.local_runner import LocalRunner
-
-
-class TestLocalRunner:
-    def test_runner_init(self):
-        r = LocalRunner()
-        assert r is not None

tests/test_midjourney.py

@@ -1,231 +0,0 @@
-"""测试 Midjourney 代理工具 — Router POST /run_tool
-
-本地服务将 JSON 原样转发至 MIDJOURNEY_API_BASE 上你已实现的三个接口：
-  POST /submit_job       cookie, prompt, user_id, mode(relaxed|fast)
-  POST /query_job_status cookie, job_id
-  POST /get_image_urls   job_id → 四张图链接
-
-用法：
-    1. tools/local/midjourney/.env：MIDJOURNEY_API_BASE
-    2. uv run python -m tool_agent
-    3. uv run python tests/test_midjourney.py
-
-端到端（可选）：设置 MIDJOURNEY_TEST_COOKIE、MIDJOURNEY_TEST_USER_ID 后脚本会
-    submit → 轮询 query → get_image_urls；否则仅校验工具已注册并退出 0。
-
-环境变量：
-    TOOL_AGENT_ROUTER_URL
-    MIDJOURNEY_SUBMIT_TOOL_ID      默认 midjourney_submit_job
-    MIDJOURNEY_QUERY_TOOL_ID       默认 midjourney_query_job_status
-    MIDJOURNEY_GET_URLS_TOOL_ID    默认 midjourney_get_image_urls
-    MIDJOURNEY_TEST_COOKIE / MIDJOURNEY_TEST_USER_ID / MIDJOURNEY_TEST_PROMPT / MIDJOURNEY_TEST_MODE
-    MIDJOURNEY_POLL_INTERVAL_S / MIDJOURNEY_POLL_MAX_WAIT_S
-"""
-
-from __future__ import annotations
-
-import io
-import os
-import sys
-import time
-from typing import Any
-
-if sys.platform == "win32":
-    _out = sys.stdout
-    if isinstance(_out, io.TextIOWrapper):
-        _out.reconfigure(encoding="utf-8")
-
-import httpx
-
-ROUTER_URL = os.environ.get("TOOL_AGENT_ROUTER_URL", "http://127.0.0.1:8001")
-T_SUBMIT = os.environ.get("MIDJOURNEY_SUBMIT_TOOL_ID", "midjourney_submit_job")
-T_QUERY = os.environ.get("MIDJOURNEY_QUERY_TOOL_ID", "midjourney_query_job_status")
-T_URLS = os.environ.get("MIDJOURNEY_GET_URLS_TOOL_ID", "midjourney_get_image_urls")
-TEST_COOKIE = os.environ.get("MIDJOURNEY_TEST_COOKIE", "").strip()
-TEST_USER_ID = os.environ.get("MIDJOURNEY_TEST_USER_ID", "").strip()
-TEST_PROMPT = os.environ.get("MIDJOURNEY_TEST_PROMPT", "a red apple on white background --v 6")
-TEST_MODE = os.environ.get("MIDJOURNEY_TEST_MODE", "fast").strip().lower()
-POLL_INTERVAL_S = float(os.environ.get("MIDJOURNEY_POLL_INTERVAL_S", "3"))
-POLL_MAX_WAIT_S = float(os.environ.get("MIDJOURNEY_POLL_MAX_WAIT_S", "600"))
-
-
-def run_tool(tool_id: str, params: dict[str, Any], timeout: float = 120.0) -> Any:
-    resp = httpx.post(
-        f"{ROUTER_URL}/run_tool",
-        json={"tool_id": tool_id, "params": params},
-        timeout=timeout,
-    )
-    resp.raise_for_status()
-    body = resp.json()
-    if body.get("status") != "success":
-        raise RuntimeError(body.get("error") or str(body))
-    result = body.get("result")
-    if isinstance(result, dict) and result.get("status") == "error":
-        raise RuntimeError(result.get("error", str(result)))
-    return result
-
-
-def _extract_job_id(data: dict[str, Any]) -> str | None:
-    if not isinstance(data, dict):
-        return None
-    for key in ("job_id", "jobId", "id", "task_id", "taskId"):
-        v = data.get(key)
-        if v is not None and str(v).strip():
-            return str(v).strip()
-    inner = data.get("data")
-    if isinstance(inner, dict):
-        return _extract_job_id(inner)
-    return None
-
-
-def _status_terminal_ok(data: dict[str, Any]) -> bool:
-    if not isinstance(data, dict):
-        return False
-    s = str(
-        data.get("status")
-        or data.get("job_status")
-        or data.get("jobStatus")
-        or data.get("state")
-        or ""
-    ).lower()
-    if not s and isinstance(data.get("data"), dict):
-        return _status_terminal_ok(data["data"])
-    return any(k in s for k in ("complete", "success", "done", "finished", "succeed", "ready"))
-
-
-def _status_terminal_fail(data: dict[str, Any]) -> bool:
-    if not isinstance(data, dict):
-        return False
-    s = str(data.get("status") or data.get("job_status") or data.get("state") or "").lower()
-    return any(k in s for k in ("fail", "error", "cancel", "canceled", "cancelled"))
-
-
-def _extract_url_list(payload: Any) -> list[str]:
-    if isinstance(payload, list):
-        return [str(x) for x in payload if isinstance(x, str) and x.startswith("http")]
-    if not isinstance(payload, dict):
-        return []
-    for key in ("image_urls", "urls", "images", "data"):
-        v = payload.get(key)
-        if isinstance(v, list):
-            out = [str(x) for x in v if isinstance(x, str) and x.startswith("http")]
-            if out:
-                return out
-        if isinstance(v, dict):
-            nested = _extract_url_list(v)
-            if nested:
-                return nested
-    return _extract_url_list(payload.get("data"))
-
-
-def main() -> None:
-    print("=" * 50)
-    print("测试 Midjourney（submit / query / get_image_urls）")
-    print("=" * 50)
-    print(f"ROUTER_URL: {ROUTER_URL}")
-
-    try:
-        r = httpx.get(f"{ROUTER_URL}/health", timeout=3)
-        print(f"Router 状态: {r.json()}")
-    except httpx.ConnectError:
-        print(f"无法连接 Router ({ROUTER_URL})，请先: uv run python -m tool_agent")
-        sys.exit(1)
-
-    print("\n--- 校验工具已注册 ---")
-    tr = httpx.get(f"{ROUTER_URL}/tools", timeout=30)
-    tr.raise_for_status()
-    tools = tr.json().get("tools", [])
-    ids = {t["tool_id"] for t in tools}
-    for tid in (T_SUBMIT, T_QUERY, T_URLS):
-        if tid not in ids:
-            print(f"错误: {tid!r} 不在 GET /tools 中。示例: {sorted(ids)[:25]}...")
-            sys.exit(1)
-        meta = next(t for t in tools if t["tool_id"] == tid)
-        print(f"  {tid}: {meta.get('name', '')} (state={meta.get('state')})")
-
-    if not TEST_COOKIE or not TEST_USER_ID:
-        print(
-            "\n未设置 MIDJOURNEY_TEST_COOKIE 与 MIDJOURNEY_TEST_USER_ID，跳过端到端；"
-            "工具注册检查已通过，退出 0。"
-        )
-        return
-
-    if TEST_MODE not in ("relaxed", "fast"):
-        print(f"错误: MIDJOURNEY_TEST_MODE 须为 relaxed 或 fast，当前: {TEST_MODE!r}")
-        sys.exit(1)
-
-    print("\n--- midjourney_submit_job ---")
-    try:
-        sub = run_tool(
-            T_SUBMIT,
-            {
-                "cookie": TEST_COOKIE,
-                "prompt": TEST_PROMPT,
-                "user_id": TEST_USER_ID,
-                "mode": TEST_MODE,
-            },
-            timeout=180.0,
-        )
-    except (RuntimeError, httpx.HTTPError) as e:
-        print(f"错误: {e}")
-        sys.exit(1)
-
-    if not isinstance(sub, dict):
-        print(f"错误: submit 返回非 object: {type(sub)}")
-        sys.exit(1)
-
-    job_id = _extract_job_id(sub)
-    if not job_id:
-        print(f"错误: 无法从 submit 响应解析 job_id: {sub}")
-        sys.exit(1)
-    print(f"job_id: {job_id}")
-
-    print("\n--- midjourney_query_job_status 轮询 ---")
-    deadline = time.monotonic() + POLL_MAX_WAIT_S
-    last: dict[str, Any] = {}
-
-    while time.monotonic() < deadline:
-        time.sleep(POLL_INTERVAL_S)
-        try:
-            q = run_tool(
-                T_QUERY,
-                {"cookie": TEST_COOKIE, "job_id": job_id},
-                timeout=120.0,
-            )
-        except (RuntimeError, httpx.HTTPError) as e:
-            print(f"轮询错误: {e}")
-            sys.exit(1)
-
-        last = q if isinstance(q, dict) else {}
-        st = last.get("status") or last.get("job_status") or last.get("state")
-        print(f"  status: {st}")
-
-        if _status_terminal_fail(last):
-            print(f"任务失败: {last}")
-            sys.exit(1)
-        if _status_terminal_ok(last):
-            break
-    else:
-        print(f"等待超时 ({POLL_MAX_WAIT_S}s)，最后响应: {last}")
-        sys.exit(1)
-
-    print("\n--- midjourney_get_image_urls ---")
-    try:
-        urls_payload = run_tool(T_URLS, {"job_id": job_id}, timeout=120.0)
-    except (RuntimeError, httpx.HTTPError) as e:
-        print(f"错误: {e}")
-        sys.exit(1)
-
-    urls = _extract_url_list(urls_payload)
-    if len(urls) < 4:
-        print(f"警告: 期望至少 4 个 http 链接，实际 {len(urls)}；原始: {str(urls_payload)[:500]}")
-        if len(urls) == 0:
-            sys.exit(1)
-
-    for i, u in enumerate(urls[:4], 1):
-        print(f"  [{i}] {u[:96]}...")
-    print("\n测试通过!")
-
-
-if __name__ == "__main__":
-    main()

tests/test_nano_banana.py

@@ -1,144 +0,0 @@
-"""测试 nano_banana — Router 调用 Gemini 图模（HTTP generateContent）
-
-前提：
-    - data/registry.json + data/sources.json 已注册 tool_id=nano_banana
-    - tools/local/nano_banana 已提供 POST /generate，且 .env 中配置 GEMINI_API_KEY
-
-用法：
-    1. uv run python -m tool_agent
-    2. uv run python tests/test_nano_banana.py
-
-模型切换（任选其一）：
-    - 不传 NANO_BANANA_MODEL：请求体不含 model，由工具侧默认（如 gemini-2.5-flash-image /
-      环境变量 GEMINI_IMAGE_MODEL）
-    - 显式切换预览图模：
-        NANO_BANANA_MODEL=gemini-3.1-flash-image-preview uv run python tests/test_nano_banana.py
-
-环境变量：
-    TOOL_AGENT_ROUTER_URL   默认 http://127.0.0.1:8001
-    NANO_BANANA_TOOL_ID     默认 nano_banana
-    NANO_BANANA_TEST_PROMPT 覆盖默认短提示词
-    NANO_BANANA_MODEL       非空时作为 params["model"] 传给 /run_tool
-"""
-
-import io
-import os
-import sys
-from typing import Any
-
-if sys.platform == "win32":
-    _out = sys.stdout
-    if isinstance(_out, io.TextIOWrapper):
-        _out.reconfigure(encoding="utf-8")
-
-import httpx
-
-ROUTER_URL = os.environ.get("TOOL_AGENT_ROUTER_URL", "http://127.0.0.1:8001")
-TOOL_ID = os.environ.get("NANO_BANANA_TOOL_ID", "nano_banana")
-NANO_BANANA_MODEL = os.environ.get("NANO_BANANA_MODEL", "").strip()
-TEST_PROMPT = os.environ.get(
-    "NANO_BANANA_TEST_PROMPT",
-    "A minimal flat icon of a yellow banana on white background, no text",
-)
-
-
-def run_tool(params: dict[str, Any], timeout: float = 180.0) -> dict[str, Any]:
-    resp = httpx.post(
-        f"{ROUTER_URL}/run_tool",
-        json={"tool_id": TOOL_ID, "params": params},
-        timeout=timeout,
-    )
-    resp.raise_for_status()
-    body = resp.json()
-    if body.get("status") != "success":
-        raise RuntimeError(body.get("error") or str(body))
-    result = body.get("result")
-    if isinstance(result, dict) and result.get("status") == "error":
-        raise RuntimeError(result.get("error", str(result)))
-    return result if isinstance(result, dict) else {}
-
-
-def _has_image_payload(data: dict[str, Any]) -> bool:
-    if not data:
-        return False
-    if data.get("images"):
-        return True
-    if data.get("image") and isinstance(data["image"], str) and len(data["image"]) > 100:
-        return True
-    if data.get("image_base64"):
-        return True
-    cands = data.get("candidates")
-    if isinstance(cands, list) and cands:
-        parts = cands[0].get("content", {}).get("parts", [])
-        for p in parts:
-            if isinstance(p, dict) and (p.get("inlineData") or p.get("inline_data")):
-                return True
-    return False
-
-
-def main():
-    print("=" * 50)
-    print("测试 nano_banana（Gemini 图模，可切换 model）")
-    print("=" * 50)
-    print(f"ROUTER_URL: {ROUTER_URL}")
-    print(f"tool_id:    {TOOL_ID}")
-    if NANO_BANANA_MODEL:
-        print(f"model:      {NANO_BANANA_MODEL}（经 params 传入）")
-    else:
-        print("model:      （未传，使用工具默认 / GEMINI_IMAGE_MODEL）")
-
-    try:
-        r = httpx.get(f"{ROUTER_URL}/health", timeout=3)
-        print(f"Router 状态: {r.json()}")
-    except httpx.ConnectError:
-        print(f"无法连接 Router ({ROUTER_URL})，请先: uv run python -m tool_agent")
-        sys.exit(1)
-
-    print("\n--- 校验工具已注册 ---")
-    tr = httpx.get(f"{ROUTER_URL}/tools", timeout=30)
-    tr.raise_for_status()
-    tools = tr.json().get("tools", [])
-    ids = {t["tool_id"] for t in tools}
-    if TOOL_ID not in ids:
-        print(f"错误: {TOOL_ID!r} 不在 GET /tools 中。当前示例: {sorted(ids)[:15]}...")
-        sys.exit(1)
-    meta = next(t for t in tools if t["tool_id"] == TOOL_ID)
-    print(f"  {TOOL_ID}: {meta.get('name', '')} (state={meta.get('state')})")
-    props = (meta.get("input_schema") or {}).get("properties") or {}
-    if "model" in props:
-        print("  input_schema 已声明 model（注册与实现应对齐）")
-    else:
-        print("  提示: input_schema 尚无 model 字段，注册表宜补充以便编排知晓可切换模型")
-
-    params: dict[str, Any] = {"prompt": TEST_PROMPT}
-    if NANO_BANANA_MODEL:
-        params["model"] = NANO_BANANA_MODEL
-
-    print("\n--- 调用生图 ---")
-    print(f"prompt: {TEST_PROMPT[:80]}{'...' if len(TEST_PROMPT) > 80 else ''}")
-
-    try:
-        data = run_tool(params, timeout=180.0)
-    except (RuntimeError, httpx.HTTPError) as e:
-        print(f"错误: {e}")
-        sys.exit(1)
-
-    print(f"\n下游返回 keys: {list(data.keys())[:20]}")
-    if rm := data.get("model"):
-        print(f"下游报告 model: {rm}")
-        if NANO_BANANA_MODEL and rm != NANO_BANANA_MODEL:
-            print(
-                f"警告: 请求 model={NANO_BANANA_MODEL!r} 与返回 model={rm!r} 不一致（若工具会规范化 ID 可忽略）"
-            )
-
-    if _has_image_payload(data):
-        print("\n检测到图片相关字段，测试通过!")
-        return
-
-    print("\n未识别到常见图片字段（images / image / candidates[].inlineData 等）。")
-    print(f"完整结果（截断）: {str(data)[:800]}")
-    sys.exit(1)
-
-
-if __name__ == "__main__":
-    main()

tests/test_registry.py

@@ -1,10 +0,0 @@
-"""测试 Registry"""
-
-import pytest
-from tool_agent.registry.registry import ToolRegistry
-
-
-class TestRegistry:
-    def test_registry_init(self):
-        r = ToolRegistry()
-        assert r.list_all() is not None

tests/test_router_agent.py

@@ -1,46 +0,0 @@
-"""测试 Router Agent 基本功能"""
-
-import asyncio
-import logging
-import sys
-
-# 修复 Windows 控制台编码问题
-if sys.platform == 'win32':
-    import io
-    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
-    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
-
-from tool_agent.router.agent import Router
-from tool_agent.router.router_agent import RouterAgent
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s"
-)
-
-async def test_router_agent():
-    """测试 Router Agent 的基本功能"""
-    print("=" * 60)
-    print("Testing Router Agent")
-    print("=" * 60)
-
-    # 初始化 Router
-    router = Router()
-    router_agent = RouterAgent(router)
-
-    # 测试 1: 搜索工具
-    print("\n[Test 1] Searching for tools...")
-    response = await router_agent.chat("列出所有可用的工具")
-    print(f"Response: {response}")
-
-    # 测试 2: 创建工具请求
-    print("\n[Test 2] Requesting tool creation...")
-    response = await router_agent.chat("我需要一个图片压缩工具")
-    print(f"Response: {response}")
-
-    print("\n" + "=" * 60)
-    print("Tests completed")
-    print("=" * 60)
-
-if __name__ == "__main__":
-    asyncio.run(test_router_agent())

tests/test_router_chat_api.py

@@ -1,74 +0,0 @@
-"""测试 Router Agent 的 HTTP 对话接口"""
-
-import asyncio
-import httpx
-import sys
-
-# 修复 Windows 控制台编码问题
-if sys.platform == 'win32':
-    import io
-    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
-    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
-
-
-async def test_chat_api():
-    """测试 Router Agent 的对话接口"""
-    base_url = "http://127.0.0.1:8001"
-
-    print("=" * 60)
-    print("Testing Router Agent Chat API")
-    print("=" * 60)
-
-    async with httpx.AsyncClient(timeout=60.0) as client:
-        # 测试 1: 健康检查
-        print("\n[Test 1] Health check...")
-        try:
-            response = await client.get(f"{base_url}/health")
-            print(f"✓ Health: {response.json()}")
-        except Exception as e:
-            print(f"✗ Health check failed: {e}")
-            return
-
-        # 测试 2: 对话 - 列出工具
-        print("\n[Test 2] Chat - List tools...")
-        try:
-            response = await client.post(
-                f"{base_url}/chat",
-                json={"message": "列出所有可用的工具"}
-            )
-            result = response.json()
-            print(f"✓ Response:\n{result.get('response', result)}")
-        except Exception as e:
-            print(f"✗ Chat failed: {e}")
-
-        # 测试 3: 对话 - 询问工具使用
-        print("\n[Test 3] Chat - Ask about tool usage...")
-        try:
-            response = await client.post(
-                f"{base_url}/chat",
-                json={"message": "image_stitcher 工具怎么用？"}
-            )
-            result = response.json()
-            print(f"✓ Response:\n{result['response']}")
-        except Exception as e:
-            print(f"✗ Chat failed: {e}")
-
-        # 测试 4: 对话 - 请求创建工具
-        print("\n[Test 4] Chat - Request tool creation...")
-        try:
-            response = await client.post(
-                f"{base_url}/chat",
-                json={"message": "我需要一个 PDF 转图片的工具"}
-            )
-            result = response.json()
-            print(f"✓ Response:\n{result['response']}")
-        except Exception as e:
-            print(f"✗ Chat failed: {e}")
-
-    print("\n" + "=" * 60)
-    print("Tests completed")
-    print("=" * 60)
-
-
-if __name__ == "__main__":
-    asyncio.run(test_chat_api())

tests/test_runcomfy_compiled_pipeline.py

@@ -1,116 +0,0 @@
-"""
-End-to-End Test for ComfyUI UI-to-API Compilation & Execution
-"""
-import json
-import base64
-import uuid
-import os
-import sys
-
-# Add src to path so we can import ComfyUICompiler
-sys.path.append(os.path.join(os.path.dirname(__file__), '../src'))
-from tool_agent.tool.comfyui_compiler import ComfyUICompiler
-
-# Import existing RunComfy utilities
-sys.path.append(os.path.join(os.path.dirname(__file__), '../tools/local/runomfy_workflow_executor'))
-import main as runcomfy_exec
-
-sys.path.append(os.path.join(os.path.dirname(__file__), '../tools/local/launch_comfy_env'))
-from comfy_launcher import launch_comfy_server
-
-sys.path.append(os.path.join(os.path.dirname(__file__), '../tools/local/runcomfy_stop_env'))
-from stop_service import stop_server
-
-def run_e2e_test():
-    UI_WORKFLOW_PATH = "tests/run_comfy/workflows/00000000-0000-0000-0000-000000001390_workflow.json"
-    OBJECT_INFO_PATH = "tests/run_comfy/object_info.json"
-    TEST_IMAGE_PATH = "control_reference.png"
-    
-    print("1. Compiling UI Workflow to API Format...")
-    with open(OBJECT_INFO_PATH, 'r', encoding='utf-8') as f:
-        obj_info = json.load(f)
-        
-    compiler = ComfyUICompiler(obj_info)
-    api_json = compiler.compile(UI_WORKFLOW_PATH)
-    print(f"Compilation Successful! Total API nodes: {len(api_json)}")
-    
-    # Check for LoadImage node and replace its image input
-    load_image_node_id = None
-    for node_id, node in api_json.items():
-        if node.get("class_type") in ["LoadImage", "VH_LoadImage"]:
-            load_image_node_id = node_id
-            break
-            
-    if not load_image_node_id:
-        print("Could not find LoadImage node in the compiled workflow!")
-        return
-        
-    print(f"2. Injecting local image '{TEST_IMAGE_PATH}' into Node {load_image_node_id}...")
-    target_filename = "e2e_test_input.png"
-    
-    if "image" in api_json[load_image_node_id]["inputs"]:
-        api_json[load_image_node_id]["inputs"]["image"] = target_filename
-    elif "image_path" in api_json[load_image_node_id]["inputs"]:
-        api_json[load_image_node_id]["inputs"]["image_path"] = target_filename
-    
-    print("3. Encoding local image to Base64...")
-    with open(TEST_IMAGE_PATH, "rb") as f:
-        img_b64 = base64.b64encode(f.read()).decode('utf-8')
-        
-    print("\n--- STAGE: DYNAMIC INFRASTRUCTURE ---")
-    active_server_id = None
-    try:
-        print("4. Launching RunComfy Cloud Machine (this takes a few minutes, please wait)...")
-        launch_info = launch_comfy_server(version_id="90f77137-ba75-400d-870f-204c614ae8a3", server_type="medium", duration=3600)
-        active_server_id = launch_info["server_id"]
-        comfy_url = launch_info["comfy_url"]
-        print(f"Machine Launched & Ready! Server ID: {active_server_id} @ {comfy_url}")
-        
-        print("\n5. Uploading injected image to RunComfy Workspace...")
-        runcomfy_exec.upload_file_from_base64(
-            comfy_url=comfy_url, 
-            filename=target_filename, 
-            base64_data=img_b64, 
-            file_type="input", 
-            subfolder=""
-        )
-        print("Image Uploaded.")
-        
-        print("6. Submitting Execution Prompt to Queue...")
-        client_id = str(uuid.uuid4())
-        prompt_id = runcomfy_exec.submit_prompt(comfy_url, api_json, client_id)
-        print(f"Prompt Submitted! Prompt ID: {prompt_id}")
-        
-        print("7. Waiting for Remote GPU Execution...")
-        runcomfy_exec.wait_for_completion(comfy_url, client_id, prompt_id, timeout=600)
-        print("Execution Completed!")
-        
-        print("8. Downloading Result Images...")
-        result_b64_list = runcomfy_exec.download_images_as_base64(comfy_url, prompt_id)
-        
-        if not result_b64_list:
-            print("No output images received.")
-        else:
-            for idx, b64_img in enumerate(result_b64_list):
-                out_name = f"tests/run_comfy/e2e_output_{idx}.jpg"
-                with open(out_name, "wb") as f:
-                    f.write(base64.b64decode(b64_img))
-                print(f"Success! Image saved to {out_name}")
-                
-    except Exception as e:
-        import requests
-        if isinstance(e, requests.exceptions.HTTPError) and e.response is not None:
-            print(f"\nPipeline Failed at Execution Stage! HTTP Error: {e.response.status_code}")
-            print(f"ComfyUI Validation Response: {e.response.text}")
-        else:
-            err_msg = str(e).encode('ascii', 'ignore').decode('ascii')
-            print(f"\nPipeline Failed at Execution Stage! Error: {err_msg}")
-    
-    finally:
-        if active_server_id:
-            print(f"\n9. Cleaning up: Stopping RunComfy Cloud Machine {active_server_id}...")
-            res = stop_server(active_server_id)
-            print(f"Machine Stop Status: {res['status']} - {res['message']}")
-
-if __name__ == "__main__":
-    run_e2e_test()

tests/test_scheduler.py

@@ -1,10 +0,0 @@
-"""测试 Scheduler"""
-
-import pytest
-from tool_agent.router.scheduler import Scheduler
-
-
-class TestScheduler:
-    def test_scheduler_init(self):
-        s = Scheduler()
-        assert s is not None

tests/test_tool_provider.py

@@ -1,40 +0,0 @@
-import os
-from dotenv import load_dotenv
-import psycopg2
-
-load_dotenv()
-
-def test_tool_provider():
-    print("Testing connection to tool_provider table...")
-    try:
-        conn = psycopg2.connect(
-            host=os.getenv('KNOWHUB_DB'),
-            port=int(os.getenv('KNOWHUB_PORT', 5432)),
-            user=os.getenv('KNOWHUB_USER'),
-            password=os.getenv('KNOWHUB_PASSWORD'),
-            database=os.getenv('KNOWHUB_DB_NAME')
-        )
-        c = conn.cursor()
-        
-        # Check if tool_provider table exists by selecting 0 rows
-        c.execute("SELECT * FROM tool_provider LIMIT 0")
-        print("Success! Table tool_provider exists.")
-        
-        # Ensure it has the correct columns
-        expected_cols = ['tool_id', 'provider_id']
-        col_names = [desc[0] for desc in c.description]
-        print(f"Columns in tool_provider: {col_names}")
-        
-        # Check if there's any data
-        c.execute("SELECT * FROM tool_provider LIMIT 5")
-        rows = c.fetchall()
-        print(f"Sample data from tool_provider: {rows}")
-        
-        cursor = conn.cursor()
-        c.close()
-        conn.close()
-    except Exception as e:
-        print(f"Failed to access tool_provider table: {e}")
-
-if __name__ == "__main__":
-    test_tool_provider()