1) 总览与选型

2) 推荐工程目录（可直接照抄）

repo/
  services/
    dispatcher/            # 统一入口：队列、路由、重试、日志、回退策略
    claude_native_worker/  # Runtime A：Claude Agent SDK（Anthropic-compatible / Poe）
    universal_worker/      # Runtime B：通用 agent loop（OpenAI-compatible / Poe）
  shared/
    tools/                 # 统一工具定义（read/search/shell/http/db...）
    sandbox/               # 隔离执行（容器/命名空间/限权/配额）
    observability/         # tracing / structured logs / artifacts
  infra/
    docker/                # Dockerfiles、compose、k8s manifests（可选）
    policies/              # tool allowlist、路径规则、网络 allowlist

3) 路线 A：Claude Agent SDK（原生） + Poe Anthropic-compatible

3.1 容器内环境变量（最小接入）

# Poe key
export POE_API_KEY="..."

# Anthropic-compatible：把 Anthropic 的变量指向 Poe
export ANTHROPIC_API_KEY="$POE_API_KEY"
export ANTHROPIC_BASE_URL="https://api.poe.com"

3.2 Claude 家族内“自动选模型”路由（示例）

• 默认平衡：claude-sonnet-4
• 快速/便宜：claude-3-5-haiku
• 高难/长推理：claude-opus-4（或同系列更强版本）

3.3 Worker 代码骨架（Python）

import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions

def pick_claude_model(task: dict) -> str:
    # 示例：你可以基于预算、延迟、难度、历史成功率等做更复杂路由
    if task.get("priority") == "fast":
        return "claude-3-5-haiku"
    if task.get("difficulty") == "hard":
        return "claude-opus-4"
    return "claude-sonnet-4"

async def run_task(task: dict) -> None:
    options = ClaudeAgentOptions(
        model=pick_claude_model(task),
        cwd=task.get("workspace", "/work"),
        allowed_tools=["Read", "Edit", "Bash", "Glob", "Grep"],
        max_turns=task.get("max_turns", 20),
        permission_mode="acceptEdits",
    )

    async for message in query(prompt=task["prompt"], options=options):
        # 事件流：写入你的日志系统 / SSE / WebSocket / queue
        print(message)

if __name__ == "__main__":
    # Poe Anthropic-compatible
    os.environ.setdefault("ANTHROPIC_API_KEY", os.environ["POE_API_KEY"])
    os.environ.setdefault("ANTHROPIC_BASE_URL", "https://api.poe.com")

    task = {"prompt": "List files and summarize TODOs", "priority": "fast"}
    asyncio.run(run_task(task))

4) A 的沙箱与权限策略（强烈建议双层防护）

4.1 沙箱层（容器/namespace）

• 只挂载 workspace：例如 /work，其余全部不可见
• 网络默认关闭：除非你明确需要 allowlist（例如仅允许访问某内部 API）
• 资源配额：CPU/内存/磁盘/进程数 + 全局超时
• 输出收集：stdout/stderr + 产物目录（artifacts）

4.2 SDK 权限层（工具级策略）

• 路径 allowlist：只能读/写 /work，禁止 /、~、/etc 等
• 命令 allowlist：只允许 ls/cat/grep/python -m ... 等，拒绝 curl/ssh/sudo 等
• 危险模式拦截：如递归删除、写入二进制可执行等

5) 路线 B：Poe OpenAI-compatible（/v1）+ 模型无关 Agent Loop

5.1 运行时配置

export POE_API_KEY="..."
# OpenAI-compatible base_url
# https://api.poe.com/v1

5.2 通用 Agent Loop（Python，最小可用）

import os, json
from openai import OpenAI
from jsonschema import validate, ValidationError

client = OpenAI(
    api_key=os.environ["POE_API_KEY"],
    base_url="https://api.poe.com/v1",
)

TOOLS = [
  {
    "type": "function",
    "function": {
      "name": "read_file",
      "description": "Read a UTF-8 text file from workspace",
      "parameters": {
        "type": "object",
        "properties": {"path": {"type": "string"}},
        "required": ["path"]
      }
    }
  }
]

def read_file(path: str) -> str:
    # 这里务必加 sandbox path 检查：只允许 /work 下
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

TOOL_IMPL = {"read_file": read_file}

def safe_tool_call(name: str, args: dict) -> str:
    schema = next(t["function"]["parameters"] for t in TOOLS if t["function"]["name"] == name)
    # Poe 兼容层可能忽略 strict / 不保证强 schema，因此必须本地验证并处理不合规
    validate(instance=args, schema=schema)
    return TOOL_IMPL[name](**args)

def run_agent(prompt: str, model: str) -> str:
    messages = [
        {"role": "system", "content": "You are a careful agent. Use tools when needed."},
        {"role": "user", "content": prompt}
    ]

    for _ in range(20):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message

        if getattr(msg, "tool_calls", None):
            for tc in msg.tool_calls:
                name = tc.function.name
                args = json.loads(tc.function.arguments or "{}")
                try:
                    out = safe_tool_call(name, args)
                    messages.append({"role": "tool", "tool_call_id": tc.id, "content": out})
                except (ValidationError, Exception) as e:
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": f"ERROR: {type(e).__name__}: {e}"
                    })
            continue

        return msg.content or ""

    return "ERROR: max turns reached"

print(run_agent("Read /work/README.md and summarize.", model="Claude-Opus-4.1"))

5.3 任意模型自动选择（建议两阶段路由）

1. 候选集筛选：按任务类型（coding/research）、预算、延迟要求、是否需要多模态等
2. 在线降级：遇到超时/429/服务繁忙 → 换模型或换 bot；必要时降级到更快更便宜的模型

6) B 的工具协议与参数校验：必做项

6.1 为什么必须本地校验？

6.2 推荐工具分层（共享给 A/B）

• Tool Spec（声明层）：JSON Schema / function signature（给模型看的）
• Tool Impl（实现层）：真正的 Python/TS 函数（做权限与路径拦截）
• Tool Runtime（执行层）：沙箱里跑、带超时、带资源限制、记录审计日志

6.3 “工具错误”也要标准化

{
  "error": true,
  "type": "ValidationError",
  "message": "path must start with /work",
  "hint": "Use /work/relative_path"
}

7) Dispatcher：统一调度、路由、回退（让两条路“都实现”的关键）

7.1 Dispatcher 必备职责（最小可用 5 件事）

1. 接收任务（HTTP / CLI / queue）
2. 路由：选择 Runtime A 或 Runtime B（或 auto）
3. 分配 workspace（每任务一个隔离目录）
4. 启动对应 worker（容器内）并收集事件流（日志/中间结果/产物）
5. 重试/回退（A 失败 → B；B 某模型失败 → 换模型）

7.2 建议路由策略（可解释、好运营）

7.3 Dispatcher 的“回退链”示例（实践里很好用）

8) 运行与部署清单（按这个顺序做最顺）

8.1 先跑通 A（Claude 原生）

1. 容器里设置：ANTHROPIC_BASE_URL=https://api.poe.com、ANTHROPIC_API_KEY=$POE_API_KEY
2. 只开只读工具：Read/Glob/Grep
3. 跑一个只读总结任务，确认事件流与输出
4. 再逐步开放 Edit/Bash，并加命令/路径策略

8.2 再跑通 B（通用 loop）

1. OpenAI SDK base_url 指向 https://api.poe.com/v1
2. 实现 2~3 个最小工具：read_file / list_dir / grep
3. 务必加 jsonschema 校验与错误回灌修复
4. 接入模型选择策略（按预算/延迟）与失败切换

8.3 最后上线 Dispatcher

1. 实现统一任务接口：/run（runtime=auto|a|b, policy=claude_family|any）
2. 每任务独立 workspace + 产物目录 + 日志 trace_id
3. 实现回退链 + 指数退避重试 + 全局超时
4. 完善观测：每回合 token/耗时/工具调用次数/错误类型分布