# 工具调用范式③:claude-code 原生车道(把工具调用外包给运行时)

← 对照卡索引
🎯 单体工作区助手套进 react-over-text 全是成本没收益(实测 76 调用 0 执行); 正解是开第二条车道:直接原生跑 claude-code,native 工具开、流式事件映射回平台 SSE。

### 📖 教科书概念

第三种范式不在平台内实现工具循环,而是把整段任务**外包**给一个已经内建了 工具执行能力的智能体运行时(这里是 claude-code CLI)。平台只负责:起进程、 喂 prompt、把运行时吐出的事件流翻译成自己的 SSE,以及划定它能动的目录与工具集。 这对应工程上"何时自己写 agent loop、何时复用现成 agent"的取舍:当任务是 "在一个文件夹里自由读写、跑命令"的单体助手时,直接调一次成熟运行时,往往优于 把它套进自家 react-over-text 封装(那是给"必须派活给子智能体"的 orchestrator 设计的)。

### 💻 平台源码落点

`lambdagent/src/lambdagent/providers/claude_code_native.py` · run_native · L127-210

    127def run_native(prompt: str, *, cwd: str, model: str = "sonnet",128               system_append: str = "", session_id: Optional[str] = None,129               on_event: Optional[Callable[[dict], None]] = None,130               config_extra: Optional[dict] = None,131               cancel=None) -> NativeResult:132    """在 cwd 里原生跑一次 claude-code,流式回调事件,返回最终结果。133134    auth(401)→ 抛 ProviderError(_AUTH_HINT)。卡死/超时 → 抛 ProviderError(retryable)。135    """136    extra = config_extra or {}137    claude_bin = _find_working_claude(extra.get("claude_bin", "claude")) or "claude"138    allowed = extra.get("native_allowed_tools") or _DEFAULT_ALLOWED_TOOLS139    strict_mcp = not bool(extra.get("native_allow_mcp", False))140    cmd = build_cmd(claude_bin, prompt, cwd=cwd, model=model,141                    system_append=system_append, session_id=session_id,142                    allowed_tools=allowed, strict_mcp=strict_mcp)143144    first_byte_timeout = float(extra.get("native_first_byte_timeout", _DEFAULT_FIRST_BYTE_TIMEOUT_S))145    idle_timeout = float(extra.get("native_idle_timeout", _DEFAULT_IDLE_TIMEOUT_S))146    hard_timeout = float(extra.get("native_hard_timeout", _DEFAULT_HARD_TIMEOUT_S))147148    logger.info("claude-code NATIVE spawn: cwd=%s model=%s resume=%s tools=%s strict_mcp=%s",149                cwd, model, bool(session_id), ",".join(allowed), strict_mcp)150151    result = NativeResult()152    t0 = time.time()153    proc = subprocess.Popen(154        cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,155        bufsize=0,  # 二进制非阻塞读,不用行缓冲(bufsize=1 在 binary 模式会告警)156    )157    buf = b""158    last_byte_at: Optional[float] = None159    err_chunks: List[bytes] = []160    saw_result = False161162    def _process_line(line: str):163        nonlocal saw_result164        line = line.strip()165        if not line:166            return167        try:168            obj = json.loads(line)169        except json.JSONDecodeError:170            return171        _handle_event(obj, on_event, result)172        if obj.get("type") == "result":173            saw_result = True174175    try:176        import fcntl177        for f in (proc.stdout, proc.stderr):178            fl = fcntl.fcntl(f, fcntl.F_GETFL)179            fcntl.fcntl(f, fcntl.F_SETFL, fl | os.O_NONBLOCK)180181        while proc.poll() is None:182            now = time.time()183            if cancel is not None and cancel.is_cancelled():184                proc.kill()185                raise cancel.CancelledRun()186            if now - t0 > hard_timeout:187                proc.kill()188                raise ProviderError(189                    f"claude-code native hard-timeout after {now-t0:.0f}s", "claude-code",190                    retryable=True)191            if last_byte_at is None and now - t0 > first_byte_timeout:192                proc.kill()193                raise ProviderError(194                    f"claude-code native first-byte-timeout after {now-t0:.0f}s", "claude-code",195                    retryable=True)196            # idle 只在没有工具在飞时判定。工具在飞(Bash 跑实验/编译可达数分钟,197            # 期间 stream-json 静默)时不杀,靠上面的 hard_timeout 兜底——否则会把198            # 正常长跑命令误判成卡死(run_84f6fdb6f11b: idle-timeout after 181s)。199            if (last_byte_at is not None and result.pending_tools == 0200                    and now - last_byte_at > idle_timeout):201                proc.kill()202                raise ProviderError(203                    f"claude-code native idle-timeout after {now-last_byte_at:.0f}s "204                    f"(no tool in flight)", "claude-code", retryable=True)205206            ready, _, _ = select.select([proc.stdout, proc.stderr], [], [], 0.5)207            for fobj in ready:208                try:209                    chunk = fobj.read()210                except (BlockingIOError, OSError):

claude -p --output-format stream-json --add-dir <文件夹> --allowedTools Read/Write/Edit/Bash/... --strict-mcp-config --permission-mode bypassPermissions;解析 stream-json 事件映射成现有 SSE。
`lambdagent/src/lambdagent/providers/claude_code_native.py` · pending_tools (idle 看门狗) · L44-110

    44    output_tokens: int = 045    cache_read_tokens: int = 046    cache_creation_tokens: int = 047    cost_usd: float = 0.048    num_turns: int = 049    is_error: bool = False50    trace: List[dict] = field(default_factory=list)51    pending_tools: int = 0   # 在飞工具数(tool_use 未等到 tool_result);>0 时禁 idle 超时525354def build_cmd(claude_bin: str, prompt: str, *, cwd: str, model: str,55              system_append: str = "", session_id: Optional[str] = None,56              allowed_tools: Optional[List[str]] = None,57              strict_mcp: bool = True) -> List[str]:58    """构造 native claude-code 命令。cwd 由调用方在 spawn 时设置。"""59    cmd = [60        claude_bin, "-p", prompt,61        "--output-format", "stream-json", "--verbose",62        "--model", model,63        # native 工具限定在工作文件夹(claude-code 的文件操作 root)。64        "--add-dir", cwd,65        # -p 非交互:必须 bypass,否则工具卡在权限审批等不到回应。66        "--permission-mode", "bypassPermissions",67        "--allowedTools", *(allowed_tools or _DEFAULT_ALLOWED_TOOLS),68    ]69    if strict_mcp:70        cmd.append("--strict-mcp-config")  # 隔离用户 MCP(Gmail/Calendar 等)71    if system_append:72        cmd += ["--append-system-prompt", system_append]73    if session_id:74        cmd += ["--resume", session_id]75    return cmd767778def _emit(on_event, etype: str, payload: dict):79    if on_event is not None:80        try:81            on_event({"event": etype, **payload})82        except Exception as e:  # 回调异常不能拖垮主流程83            logger.warning("native on_event(%s) failed: %s", etype, e)848586def _handle_event(obj: dict, on_event, result: NativeResult):87    """把一条 claude stream-json 事件映射成平台 SSE 事件 + 累积结果。"""88    t = obj.get("type")89    if t == "system" and obj.get("subtype") == "init":90        result.session_id = obj.get("session_id") or result.session_id91        return92    if t == "assistant":93        for c in (obj.get("message", {}) or {}).get("content", []) or []:94            if not isinstance(c, dict):95                continue96            if c.get("type") == "text" and c.get("text"):97                _emit(on_event, "think_chunk", {"text": c["text"]})98            elif c.get("type") == "tool_use":99                result.pending_tools += 1   # 工具开飞 → 禁 idle 超时直到结果回来100                _emit(on_event, "tool_call",101                      {"tool": c.get("name", ""), "input": c.get("input", {})})102                result.trace.append({"tool": c.get("name", ""),103                                     "input": str(c.get("input", ""))[:300]})104        return105    if t == "user":106        for c in (obj.get("message", {}) or {}).get("content", []) or []:107            if isinstance(c, dict) and c.get("type") == "tool_result":108                result.pending_tools = max(0, result.pending_tools - 1)109                out = c.get("content", "")110                if isinstance(out, list):

坑:长命令(实验/pdflatex)期间 stream-json 在工具返回前完全静默,会被 idle 看门狗误杀。修法:跟踪 pending_tools(tool_use+1 / tool_result-1),idle 仅在 pending_tools==0 时生效,在飞工具靠 hard_timeout 兜底。

### ∑ 形式化对象

外包算子 RunNative(cwd, allowedTools) : Task → (Effects, Result)
effect: IO[fs, shell] ⊗ Cap(allowedTools, cwd)

与①②不同,这里的 effect 不是平台逐步推断的,而是把一整束能力(读写文件、跑 shell) **以能力集 Cap 的形式**授予外部运行时,并用 cwd + allowedTools 划界。形式化关注点 从"每步 effect"转为"授权边界是否正确"——这正好引出 week 8 沙箱/能力的讨论。

### 🎞 Live Demo

对照实验同样进对比台。已 live 验证:真起 claude -p 在临时目录写出文件, 产出 tool_call / tool_result 事件,cost/session 正确。注意路由按 provider 分流—— 切到 qwen/ollama 必须退回 react 车道(claude -p 只认 sonnet/opus/haiku)。

学习目标 / Takeaway 三范式并排,学生应能在"自己写 loop 的精细控制"与"外包给成熟运行时的省心"之间 做权衡判断,并说清:范式③把 effect 推断换成了能力授权边界(Cap), 可靠性问题(静默/卡死/路由)随之从"解析正确性"转移到"进程与授权管理"。

相关对照卡: 工具调用范式①:react-over-text(把工具调用当散文手写 JSON) 工具调用范式②:原生 function-calling(结构化 tool_calls) 硬沙箱:把'写文件'这一 effect 限制在能力边界内