CLI调用Claude做产品级服务我试过翻车了去年我接过一个活把Claude Code CLI包成HTTP接口给上层业务调用subprocess起进程、stdin塞prompt、stdout收文本简单粗暴。第一周跑得欢第二周产品提需求能不能记住上一次问的内容能不能限定只读能不能输出JSON我一一在CLI外面糊补丁越糊越脏最后代码长得像一碗意大利面。后来Anthropic出了Agent SDK我把那堆subprocess代码全删了三百行Python搞定。这一篇就讲怎么用Agent SDK把Claude塞进Python Web服务。书里对应第8章M7里程碑——控制粒度演进的终点CLAUDE.md声明式→ Skills半声明式→ Hooks事件式→Agent SDK编程式。前三层还在配置范畴到SDK这一层Harness本身被当成一个库来调用。从工具到组件SDK的定位CLI是工具SDK是组件。工具靠shell调用、靠stdin/stdout通信组件靠import、靠对象方法通信。前者面向人后者面向代码。# Pythonpipinstallclaude-agent-sdk# TypeScript/Node.jsnpminstallanthropic-ai/claude-agent-sdk两个语言版本API基本对齐下面以Python为主。Quick Start五分钟跑起来importasynciofromclaude_agent_sdkimportquery,ClaudeAgentOptionsasyncdefanalyze_code():optionsClaudeAgentOptions(max_turns5,allowed_tools[Read,Grep,Glob],system_prompt你是一名代码架构分析师。,)asyncformessageinquery(prompt分析 src/auth/ 目录的实现架构,optionsoptions):ifmessage.typeassistant:forblockinmessage.content:ifhasattr(block,text):print(block.text,end,flushTrue)elifmessage.typeresult:print(f\n\n完成。费用${message.total_cost_usd:.4f})asyncio.run(analyze_code())query()是异步生成器吐出来的不是一坨字符串是结构化消息。TS版本几乎一模一样把async for换成for await、把hasattr换成text in block即可。等一下这里我漏说一个前提——query()吐的消息不止两种一共四类搞不清楚状态机后面会迷糊。四种消息类型对话状态机// 1. system_init会话初始化给 session_id{type:system,subtype:init,session_id:550e8400-...,model:claude-sonnet-4-6,tools:[Read,Grep,Glob]}// 2. assistantClaude 的响应既能有文本又能有 tool_use{type:assistant,message:{role:assistant,content:[{type:text,text:让我先看看目录结构...},{type:tool_use,id:toolu_xxx,name:Glob,input:{pattern:src/auth/**/*}}]}}// 3. user工具执行结果回灌{type:user,message:{role:user,content:[{type:tool_result,tool_use_id:toolu_xxx,content:src/auth/\n├── login.ts\n├── session.ts}]}}// 4. result任务完成带成本/耗时/session_idsystem_init给session_idassistant里既能有文本又能有tool_useuser是工具结果回灌result收尾给费用和元数据。我第一次写的时候把result当成了最终答案漏掉了assistant流里也可能有最终文本结果输出残缺——记住最终文本在assistant流里result只是元数据。ClaudeAgentOptions精细控制从这开始CLI时代我想要的控制项这里都有fromclaude_agent_sdkimportClaudeAgentOptions optionsClaudeAgentOptions(modelclaude-sonnet-4-6,max_turns10,max_budget_usd1.0,# 成本上限超了就停allowed_tools[Read,Grep,Glob,Write],disallowed_tools[Bash],permission_modedefault,# default / acceptEdits / plan / bypassPermissionssystem_prompt你是一名高级代码审查员。,append_system_prompt务必检查 SQL 注入漏洞。,# 追加不覆盖cwd/path/to/project,env{PROJECT_NAME:MyApp},resumesession-id-to-resume,# 续接会话output_format{type:json_schema,schema:my_schema},mcp_servers[{name:db,command:python,args:[./db_server.py]}],)max_budget_usd这条我必须有——给客户做项目的时候没有成本上限的Agent能在测试循环里把预算烧光亲身经历一个晚上烧了四十刀。工具权限还能做模式匹配颗粒度到命令参数optionsClaudeAgentOptions(allowed_tools[Read,Grep,Glob,Bash(git diff *),# 只允许 git diffBash(npm test *),# 只允许 npm testmcp__database__query,# 只允许这个 MCP 工具])Bash(git diff *)这种写法比disallowed_tools[Bash]然后自己写正则过滤命令行优雅多了——CLI时代我就是这么糊的这里是SDK原生支持。session_id会话延续和分支# 第一轮分析问题拿到 session_idsession_idNoneasyncformessageinquery(prompt分析 src/auth 的安全问题,optionsoptions):ifmessage.typesystemandmessage.subtypeinit:session_idmessage.session_id# 第二轮在上一轮上下文里继续resume_optionsClaudeAgentOptions(**options.__dict__,resumesession_id)asyncformessageinquery(prompt重点分析你发现的第一个 SQL 注入风险,optionsresume_options):...要分支探索两个方向加fork_sessionTrueoptions_forkClaudeAgentOptions(**options.__dict__,resumesession_id,fork_sessionTrue)# 方向 A重构为微服务asyncformessageinquery(prompt如果重构为微服务需要改哪些,optionsoptions_fork):...# 方向 B原架构加固同一个 session_id 再 forkasyncformessageinquery(prompt如果保持现有架构怎么加固安全,optionsoptions_fork):...fork_session不污染原会话——做A/B方案对比的时候特别好用。tool 装饰器自定义工具 Pydanticfromclaude_agent_sdkimporttool,create_sdk_mcp_serverfrompydanticimportBaseModel,FieldclassDatabaseQueryParams(BaseModel):table:strField(...,descriptionTable name)columns:list[str]Field(default[*],descriptionColumns to select)where:str|NoneField(defaultNone,descriptionWHERE clause)limit:intField(default100,ge1,le1000)# 1到1000之间tool(namesafe_query,descriptionExecute a safe, parameterized database query,parametersDatabaseQueryParams# 直接传 Pydantic 类)asyncdefsafe_query(args:DatabaseQueryParams):# args 已经通过 Pydantic 验证类型安全rowsawaitdb.execute(args.table,args.columns,args.where,args.limit)return{content:[{type:text,text:json.dumps(rows)}]}# 用 MCP 服务器承载这些工具tools_servercreate_sdk_mcp_server(nameapp-tools,version1.0.0,tools[safe_query])optionsClaudeAgentOptions(mcp_servers{app-tools:tools_server},allowed_tools[Read,Grep,Glob,mcp__app-tools__safe_query])parameters直接传Pydantic类SDK会自动转成JSON Schema喂给ClaudeClaude回传的参数也会被Pydantic校验——limit超1000直接拒。CLI时代我得自己写参数校验还经常漏边界。等一下这里我又漏了一个前提——光有参数校验不够还得有运行时拦截。四道安全防线脱离CLI沙箱也得住CLI有沙箱兜底SDK脱离了CLI得自己搭防线。书上给了四道fromclaude_agent_sdkimportClaudeAgentOptions,HookMatcher# 防线一: PreToolUse Hook——执行前拦截asyncdefblock_dangerous_bash(input_data,tool_use_id,context):ifinput_data[tool_name]!Bash:return{}commandinput_data[tool_input].get(command,)dangerous[rm -rf,sudo,chmod 777, /dev/,mkfs,dd if]forpatternindangerous:ifpatternincommand:return{hookSpecificOutput:{hookEventName:PreToolUse,permissionDecision:deny,permissionDecisionReason:fBlocked:{pattern}}}return{}# 防线二: can_use_tool——运行时权限检查asyncdefcan_use_tool(tool_name:str,tool_input:dict)-dict:iftool_namein[Write,Edit]:file_pathtool_input.get(file_path,)if.envinfile_pathorsecretsinfile_path:return{allowed:False,reason:Access to sensitive files denied}iftool_nameBash:commandtool_input.get(command,)ifany(cmdincommandforcmdin[curl,wget,ssh]):return{allowed:False,reason:Network commands not allowed}return{allowed:True}# 防线三: PostToolUse——执行后审计asyncdefaudit_all_tools(input_data,tool_use_id,context):importjsonfromdatetimeimportdatetime entry{timestamp:datetime.now().isoformat(),tool:input_data[tool_name],input:input_data[tool_input],}withopen(agent-audit.jsonl,a)asf:f.write(json.dumps(entry)\n)return{}optionsClaudeAgentOptions(permission_modeacceptEdits,# 防线零: 权限模式allowed_tools[Read,Write,Edit,Grep,Glob],# 防线一: 工具白名单can_use_toolcan_use_tool,# 防线二: 运行时检查hooks{# 防线三: Hooks 拦截审计PreToolUse:[HookMatcher(matcherBash,hooks[block_dangerous_bash])],PostToolUse:[HookMatcher(matcher*,hooks[audit_all_tools])],})四道防线permission_mode粗粒度模式allowed_tools工具白名单can_use_tool运行时检查PreToolUse/PostToolUseHooks事件拦截审计。我做生产服务时这四道全开少一道都睡不着。结构化输出强制JSON SchemafrompydanticimportBaseModelclassSecurityReport(BaseModel):summary:strissues:list[dict]# [{severity, file, line, description}]risk_score:float# 0.0 - 10.0optionsClaudeAgentOptions(output_format{type:json_schema,schema:SecurityReport.model_json_schema()},max_turns10,allowed_tools[Read,Grep,Glob],)asyncformessageinquery(prompt对 src/ 进行安全审查,optionsoptions):ifmessage.typeresultandmessage.structured_output:reportSecurityReport.model_validate(message.structured_output)print(f风险评分{report.risk_score})forissueinreport.issues:print(f [{issue[severity]}]{issue[file]}:{issue.get(line,?)})message.structured_output是SDK帮你validate好的dict再用Pydantic包一层就是类型安全对象。CLI时代我用正则解析Claude的Markdown输出写到崩溃。完整Web服务FastAPI SSE把上面这些拼起来一个能跑的代码分析服务#!/usr/bin/env python3代码分析 Agent 服务——一个可运行的完整示例importasyncio,jsonfromclaude_agent_sdkimportquery,ClaudeAgentOptionsasyncdefanalyze_codebase(directory:str,focus:strgeneral):focus_prompts{security:专注于安全漏洞SQL 注入、XSS、敏感信息硬编码、权限控制。,performance:专注于性能问题N1 查询、内存泄漏、缺少缓存。,quality:专注于代码质量命名规范、DRY 原则、复杂度、测试覆盖。,general:全面分析安全、性能、质量、架构。}optionsClaudeAgentOptions(modelclaude-sonnet-4-6,max_turns15,max_budget_usd0.50,allowed_tools[Read,Grep,Glob],permission_modeplan,# 只读模式cwddirectory,append_system_promptfocus_prompts.get(focus,focus_prompts[general]),)output_text,tools_used,metadata[],[],{}asyncformessageinquery(promptf分析当前项目的代码。输出 Markdown 格式的分析报告。,optionsoptions,):ifmessage.typeassistant:forblockinmessage.content:ifhasattr(block,text):output_text.append(block.text)elifhasattr(block,name):tools_used.append(block.name)elifmessage.typeresult:metadata{session_id:message.session_id,cost_usd:message.total_cost_usd,turns:message.num_turns,duration_ms:message.duration_ms,success:notmessage.is_error,}return{report:\n.join(output_text),tools_used:tools_used,metadata:metadata}包成FastAPI接口加SSE流式输出fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponse appFastAPI()app.post(/api/analyze)asyncdefanalyze(request:AnalyzeRequest):asyncdefevent_stream():asyncformessageinquery(promptrequest.prompt,optionsoptions):ifmessage.typeassistant:forblockinmessage.content:ifhasattr(block,text):yieldfdata:{json.dumps({type:text,content:block.text})}\n\nelifmessage.typeresult:yieldfdata:{json.dumps({type:done,cost:message.total_cost_usd})}\n\nreturnStreamingResponse(event_stream(),media_typetext/event-stream)StreamingResponse把async for的每一段文本立刻推给前端用户体验比等三十秒再一次性返回好太多。顺便一提我自己的雷达鸭App华为应用市场微信小程序收录中国一人公司赚钱案例的AI分析接口也是用Agent SDK包成FastAPI服务挂在内网前端Uni-appArkTS直接SSE消费比之前subprocess那套稳定多了。下一版我想要什么我对下一版SDK有两个期待一是can_use_tool支持异步流式审批——现在一次只能返回allow/deny复杂策略要写一堆if-else二是output_format能原生支持流式部分JSON——现在结构化输出必须等result才能拿到完整对象长报告场景下用户得干等。Agent SDK把Harness从开发者用的工具变成了产品里的组件控制粒度到命令参数、会话分支、运行时拦截这一层。CLI的活本来就不该让产品服务去干。关于作者雷达鸭App独立开发者10年软件开发经验软件设计师人工智能应用工程师专注鸿蒙ArkTSWeb前端探索AI自动化。版权声明本文基于《Claude Code 实战Harness 工程之道》黄佳 著第8章内容整理原书内容版权归原作者所有。本文采用 MIT 协议发布转载请保留本声明。