← 对照卡索引
模块 A第 2 周 tool-claude-code-native

工具调用范式③: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,流式回调事件,返回最终结果。
133
134    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_TOOLS
139    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)
143
144    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))
147
148    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)
150
151    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] = None
159    err_chunks: List[bytes] = []
160    saw_result = False
161
162    def _process_line(line: str):
163        nonlocal saw_result
164        line = line.strip()
165        if not line:
166            return
167        try:
168            obj = json.loads(line)
169        except json.JSONDecodeError:
170            return
171        _handle_event(obj, on_event, result)
172        if obj.get("type") == "result":
173            saw_result = True
174
175    try:
176        import fcntl
177        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)
180
181        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 == 0
200                    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)
205
206            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 = 0
45    cache_read_tokens: int = 0
46    cache_creation_tokens: int = 0
47    cost_usd: float = 0.0
48    num_turns: int = 0
49    is_error: bool = False
50    trace: List[dict] = field(default_factory=list)
51    pending_tools: int = 0   # 在飞工具数(tool_use 未等到 tool_result);>0 时禁 idle 超时
52
53
54def 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 cmd
76
77
78def _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)
84
85
86def _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_id
91        return
92    if t == "assistant":
93        for c in (obj.get("message", {}) or {}).get("content", []) or []:
94            if not isinstance(c, dict):
95                continue
96            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        return
105    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 限制在能力边界内