企业官网建设流程全解析

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.jsnpminstall@anthropic-ai/claude-agent-sdk

两个语言版本API基本对齐，下面以Python为主。

Quick Start：五分钟跑起来

importasynciofromclaude_agent_sdkimportquery,ClaudeAgentOptionsasyncdefanalyze_code():options=ClaudeAgentOptions(max_turns=5,allowed_tools=["Read","Grep","Glob"],system_prompt="你是一名代码架构分析师。",)asyncformessageinquery(prompt="分析 src/auth/ 目录的实现架构",options=options):ifmessage.type=="assistant":forblockinmessage.content:ifhasattr(block,'text'):print(block.text,end="",flush=True)elifmessage.type=="result":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. assistant：Claude 的响应，既能有文本又能有 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_id

system_init给session_id，assistant里既能有文本又能有tool_use，user是工具结果回灌，result收尾给费用和元数据。我第一次写的时候把result当成了"最终答案"，漏掉了assistant流里也可能有最终文本，结果输出残缺——记住，最终文本在assistant流里，result只是元数据。

ClaudeAgentOptions：精细控制从这开始

CLI时代我想要的控制项，这里都有：

fromclaude_agent_sdkimportClaudeAgentOptions options=ClaudeAgentOptions(model="claude-sonnet-4-6",max_turns=10,max_budget_usd=1.0,# 成本上限，超了就停allowed_tools=["Read","Grep","Glob","Write"],disallowed_tools=["Bash"],permission_mode="default",# default / acceptEdits / plan / bypassPermissionssystem_prompt="你是一名高级代码审查员。",append_system_prompt="务必检查 SQL 注入漏洞。",# 追加不覆盖cwd="/path/to/project",env={"PROJECT_NAME":"MyApp"},resume="session-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能在测试循环里把预算烧光，亲身经历，一个晚上烧了四十刀。

工具权限还能做模式匹配，颗粒度到命令参数：

options=ClaudeAgentOptions(allowed_tools=["Read","Grep","Glob","Bash(git diff *)",# 只允许 git diff"Bash(npm test *)",# 只允许 npm test"mcp__database__query",# 只允许这个 MCP 工具])

Bash(git diff *)这种写法比disallowed_tools=["Bash"]然后自己写正则过滤命令行优雅多了——CLI时代我就是这么糊的，这里是SDK原生支持。

session_id：会话延续和分支

# 第一轮：分析问题，拿到 session_idsession_id=Noneasyncformessageinquery(prompt="分析 src/auth 的安全问题",options=options):ifmessage.type=="system"andmessage.subtype=="init":session_id=message.session_id# 第二轮：在上一轮上下文里继续resume_options=ClaudeAgentOptions(**options.__dict__,resume=session_id)asyncformessageinquery(prompt="重点分析你发现的第一个 SQL 注入风险",options=resume_options):...

要分支探索两个方向，加fork_session=True：

options_fork=ClaudeAgentOptions(**options.__dict__,resume=session_id,fork_session=True)# 方向 A：重构为微服务asyncformessageinquery(prompt="如果重构为微服务，需要改哪些？",options=options_fork):...# 方向 B：原架构加固，同一个 session_id 再 forkasyncformessageinquery(prompt="如果保持现有架构，怎么加固安全？",options=options_fork):...

fork_session不污染原会话——做A/B方案对比的时候特别好用。

@tool 装饰器：自定义工具 + Pydantic

fromclaude_agent_sdkimporttool,create_sdk_mcp_serverfrompydanticimportBaseModel,FieldclassDatabaseQueryParams(BaseModel):table:str=Field(...,description="Table name")columns:list[str]=Field(default=["*"],description="Columns to select")where:str|None=Field(default=None,description="WHERE clause")limit:int=Field(default=100,ge=1,le=1000)# 1到1000之间@tool(name="safe_query",description="Execute a safe, parameterized database query",parameters=DatabaseQueryParams# 直接传 Pydantic 类)asyncdefsafe_query(args:DatabaseQueryParams):# args 已经通过 Pydantic 验证，类型安全rows=awaitdb.execute(args.table,args.columns,args.where,args.limit)return{"content":[{"type":"text","text":json.dumps(rows)}]}# 用 MCP 服务器承载这些工具tools_server=create_sdk_mcp_server(name="app-tools",version="1.0.0",tools=[safe_query])options=ClaudeAgentOptions(mcp_servers={"app-tools":tools_server},allowed_tools=["Read","Grep","Glob","mcp__app-tools__safe_query"])

parameters直接传Pydantic类，SDK会自动转成JSON Schema喂给Claude，Claude回传的参数也会被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{}command=input_data["tool_input"].get("command","")dangerous=["rm -rf","sudo","chmod 777","> /dev/","mkfs","dd if="]forpatternindangerous:ifpatternincommand:return{"hookSpecificOutput":{"hookEventName":"PreToolUse","permissionDecision":"deny","permissionDecisionReason":f"Blocked:{pattern}"}}return{}# 防线二: can_use_tool——运行时权限检查asyncdefcan_use_tool(tool_name:str,tool_input:dict)->dict:iftool_namein["Write","Edit"]:file_path=tool_input.get("file_path","")if".env"infile_pathor"secrets"infile_path:return{"allowed":False,"reason":"Access to sensitive files denied"}iftool_name=="Bash":command=tool_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{}options=ClaudeAgentOptions(permission_mode="acceptEdits",# 防线零: 权限模式allowed_tools=["Read","Write","Edit","Grep","Glob"],# 防线一: 工具白名单can_use_tool=can_use_tool,# 防线二: 运行时检查hooks={# 防线三: Hooks 拦截+审计"PreToolUse":[HookMatcher(matcher="Bash",hooks=[block_dangerous_bash])],"PostToolUse":[HookMatcher(matcher="*",hooks=[audit_all_tools])],})

四道防线：permission_mode（粗粒度模式）+allowed_tools（工具白名单）+can_use_tool（运行时检查）+PreToolUse/PostToolUseHooks（事件拦截+审计）。我做生产服务时这四道全开，少一道都睡不着。

结构化输出：强制JSON Schema

frompydanticimportBaseModelclassSecurityReport(BaseModel):summary:strissues:list[dict]# [{severity, file, line, description}]risk_score:float# 0.0 - 10.0options=ClaudeAgentOptions(output_format={"type":"json_schema","schema":SecurityReport.model_json_schema()},max_turns=10,allowed_tools=["Read","Grep","Glob"],)asyncformessageinquery(prompt="对 src/ 进行安全审查",options=options):ifmessage.type=="result"andmessage.structured_output:report=SecurityReport.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:str="general"):focus_prompts={"security":"专注于安全漏洞：SQL 注入、XSS、敏感信息硬编码、权限控制。","performance":"专注于性能问题：N+1 查询、内存泄漏、缺少缓存。","quality":"专注于代码质量：命名规范、DRY 原则、复杂度、测试覆盖。","general":"全面分析：安全、性能、质量、架构。"}options=ClaudeAgentOptions(model="claude-sonnet-4-6",max_turns=15,max_budget_usd=0.50,allowed_tools=["Read","Grep","Glob"],permission_mode="plan",# 只读模式cwd=directory,append_system_prompt=focus_prompts.get(focus,focus_prompts["general"]),)output_text,tools_used,metadata=[],[],{}asyncformessageinquery(prompt=f"分析当前项目的代码。输出 Markdown 格式的分析报告。",options=options,):ifmessage.type=="assistant":forblockinmessage.content:ifhasattr(block,'text'):output_text.append(block.text)elifhasattr(block,'name'):tools_used.append(block.name)elifmessage.type=="result":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 app=FastAPI()@app.post("/api/analyze")asyncdefanalyze(request:AnalyzeRequest):asyncdefevent_stream():asyncformessageinquery(prompt=request.prompt,options=options):ifmessage.type=="assistant":forblockinmessage.content:ifhasattr(block,'text'):yieldf"data:{json.dumps({'type':'text','content':block.text})}\n\n"elifmessage.type=="result":yieldf"data:{json.dumps({'type':'done','cost':message.total_cost_usd})}\n\n"returnStreamingResponse(event_stream(),media_type="text/event-stream")

StreamingResponse把async for的每一段文本立刻推给前端，用户体验比"等三十秒再一次性返回"好太多。

顺便一提，我自己的雷达鸭App（华为应用市场+微信小程序，收录中国一人公司赚钱案例）的AI分析接口，也是用Agent SDK包成FastAPI服务挂在内网，前端Uni-app+ArkTS直接SSE消费，比之前subprocess那套稳定多了。

下一版我想要什么

我对下一版SDK有两个期待：一是can_use_tool支持异步流式审批——现在一次只能返回allow/deny，复杂策略要写一堆if-else；二是output_format能原生支持流式部分JSON——现在结构化输出必须等result才能拿到完整对象，长报告场景下用户得干等。

Agent SDK把Harness从"开发者用的工具"变成了"产品里的组件"，控制粒度到命令参数、会话分支、运行时拦截这一层。CLI的活，本来就不该让产品服务去干。

关于作者：雷达鸭App独立开发者，10+年软件开发经验，软件设计师，人工智能应用工程师，专注鸿蒙ArkTS+Web前端，探索AI自动化。

版权声明：本文基于《Claude Code 实战：Harness 工程之道》（黄佳 著）第8章内容整理，原书内容版权归原作者所有。本文采用 MIT 协议发布，转载请保留本声明。