在智能体应用还停留在“谁的客户端就配谁的私有接口”那个阶段时我第一次看到 Agent Client Protocol 这个词第一反应是这不就是把 HTTP 又包了一层吗等我真的把协议规范啃完、又在本地搭了一个服务端样例跑通之后我改变了自己的判断。Agent Client Protocol后面统称 ACP解决的是智能体应用里一个非常具体但一直没人统一的痛点前端应用和后端智能体服务之间到底用什么方式说话、怎么提交任务、怎么回传工具调用结果、怎么在需要用户确认时停下来等人。简单说ACP 是一套面向智能体客户端与服务端的应用层通信规范。它定义了任务如何创建、会话如何管理、交互轮次如何推进、工具调用结果如何回传、权限请求如何审批。如果你正在做带工具调用的应用、多步骤任务执行器或者是多智能体协作的编排系统这套协议能帮我们把前端的展示逻辑和后端的执行逻辑彻底解耦省掉自己发明一套私有协议的大量弯路。这篇文章我会从协议动机、核心消息结构、服务端实操搭建、落地避坑四个角度拆着讲适合想直接照着抄作业的开发者也适合想搞清楚“为什么需要这个东西”的架构师。1. 为什么智能体应用需要一套“客户端—服务端”协议1.1 ACP 要解决的三个具体痛点先说第一个痛点私有协议太多。我们团队最早做任务型智能体时客户端和服务端的通信方式是“今天想怎么传就怎么传”。前端页面直接把一堆 prompt 拼好甩给后端模型服务后端返回一段 JSON里面塞满原始文本、中间步骤、工具返回值字段命名还经常变。这让前端页面和后端逻辑紧紧耦合在一起想换一个模型厂商或者换一套工具执行器就得同步改两端接口。ACP 要做的就是把这层交互标准化客户端只负责展示和输入服务端只负责执行和推进两边按照约定好的消息结构和状态机通信。第二个痛点是权限交互难做。真实智能体执行任务时不是所有操作都能自动完成。工具要发一封邮件、要改线上配置、要删一个文件这些高风险动作通常需要用户确认。没有协议约定的时候我们只能用“任务先挂起然后前端轮询某个状态字段”这种笨办法。轮询间隔短了浪费资源长了用户体验又差。ACP 把权限请求设计成一种显式消息服务端主动向客户端发起 permission_request客户端反馈 approve 或 deny整个流程有标准状态可追踪体验完全不一样。第三个痛点是状态同步和可观测性弱。前端拿到一个任务之后它处于 pending、working、needs_input还是已经 failed中间产出了哪些附件前几步又做了哪些工具调用这些信息如果全靠接口文档一张嘴排查问题时两眼一抹黑。ACP 定义了一套稳定的任务状态机和标准事件流服务端状态一变就通过 SSE 推送事件前端订阅事件做响应式更新不需要每秒钟拉一次接口。对做可视化工作台、聊天界面、审计系统的团队来说这个标准化价值非常大。1.2 ACP 与“模型上下文协议”的定位差异说到这里很多人会自然想到现在社区里热门的 MCPModel Context Protocol这类面向模型和工具之间上下文交换的协议。它们之间不是替代关系更多是互补。可以把 MCP 理解成“模型外挂工具的标准插座”。模型要查数据、要读写文件、要调用业务系统都是通过 MCP 去发现工具、传递上下文、拿回工具结果。它解决的是模型这一层如何连接外部能力的问题。而 ACP 解决的是“一个应用客户端如何把一个任务委托给智能体服务并持续跟踪处理过程”的问题。它更偏业务协作流任务谁发起、状态谁维护、权限谁审批、结果谁接收。在实际系统里两者可以同时存在。我去年做的一个模拟项目 X客户端是一个对话工作台它通过 ACP 把“帮我准备月度报表”这个任务提交给智能体服务端服务端内部再用 MCP 调数据库工具、文件聚合工具、导出工具。前端完全不需要知道后端到底调用了哪些工具这就是分层带来的价值。1.3 为什么不能“直接用 WebSocket 随便做”一定要争辩的话WebSocket 当然可以做类似的通信。但关键问题是WebSocket 只给了一个全双工通道它不管消息格式、错误码、权限流程、任务生命周期这些东西。你还是得自己定义一套提交流程、一套状态机、一套重试约定。ACP 相当于把这些已经踩过无数遍坑的设计整理成了现成的规范。你可以把它理解成“智能体世界的 HTTP”——HTTP 本身也不实现业务逻辑但它定义了方法、状态码、头部这些基础知识所有人都按这个来生态才能互通。所以我的建议很明确如果只是自己内部联调随便怎么玩都行如果想让前端应用能够对接多个智能体服务、甚至未来开源出去让别人接入早点拥抱 ACP 这种标准协议比后期再做适配成本低得多。2. Agent Client Protocol 的核心消息结构与状态机2.1 任务生命周期与五类核心消息ACP 的核心对象是“任务”。一个任务从创建到结束会经过几个稳定的状态我用下面的表来整理状态含义常见后续动作pending任务已创建服务端排队或正在准备进入 workingworking服务端正在规划、调用工具、生成内容完成后进入 completed出错进入 failed需要人工时进入 needs_inputneeds_input服务端需要客户端提供附加消息或审批权限客户端提交消息后回到 workingcompleted任务正常结束产出物可获取客户端读取 outputscanceled任务被取消无failed执行失败带错误信息客户端可以展示原因并决定重试围绕任务的生命周期ACP 的消息大致可以分成五类第一类是任务创建与查询消息客户端创建任务、获取任务列表、获取单个任务信息第二类是交互消息客户端向任务提交内容服务端返回当前状态和新的交互轮次第三类是权限请求消息服务端请求执行高风险操作前向客户端发送审批请求第四类是事件流消息服务端通过 SSE 推送任务更新、权限等待、错误通知等事件第五类是产出物消息任务完成后客户端拉取最终产物。这五类消息构成了一个相对完整的闭环。我最开始看协议时觉得字段有点多但真正实现一遍后发现每一个字段都有对应场景没有交互轮次字段权限审批和补充提问就无处安放没有输出引用字段附件这类二进制产物就只能靠不规范的路由传递。2.2 交互消息的字段设计与参数选择一个典型的任务创建请求通常会包含任务标题、任务描述、可选的元信息。响应里最重要的字段是任务 ID 和当前状态。下面是我在本地复刻时用的简化结构// 创建任务请求 POST /acp/tasks/create { title: 生成月度销售报告, description: 汇总最近30天销售数据输出PDF文件, metadata: { from_user: u_1024, tenant_id: t_abc } }// 创建任务响应 { id: task_8f3a2c1, status: working, interactive_turns: [], created_at: 2025-05-20T10:00:00Z }在交互消息中客户端通过 submit_message 提交新内容内容里除了文本还可以有附件引用、审批结果等。我建议在设计字段时注意三点一是任务 ID 要全局唯一并且最好带业务前缀。我们当时用“task_”开头排查日志的时候 grep 非常方便单纯用纯数字序列号很容易和数据库主键混淆。二是状态字段要保持稳定枚举。不要在客户端用“0、1、2”这种魔法数字更不要私自加类似“almost_done”这种非标准状态。所有状态尽可能用字符串枚举这样前端、服务端、日志三方都能一眼读懂。三是交互轮次数组要完整记录每一轮请求和响应。前端界面需要把“用户输入—工具执行过程—权限请求—最终结果”完整呈现出来交互轮次就是数据基础。如果嫌全量传递太重可以用增量事件和游标机制但至少要保证任务详情接口能够按需回放。权限请求消息是我认为整个协议里最有价值的部分。它的结构一般是请求 ID、请求类型、请求目标、可选的上下文说明、超时时间。客户端收到后会以请求 ID 为准答复 approve 或 deny{ request_id: req_5e6f, action: send_email, target: { recipient: opsexample.com, subject: 线上配置变更通知 }, rationale: 任务需要通知运维人员确认变更完成, timeout_ms: 300000 }这里有个细节权限请求最好带一个超时时间客户端如果长时间没给答复服务端不能无限等下去。超时后建议自动降级为“默认拒绝”并在任务状态里记录一条超时事件避免出现“任务卡在 needs_input 一辈子”的情况。2.3 服务发现、鉴权与会话标识要让客户端一接入就能找到服务端协议通常建议通过 well-known 路径做服务发现。本地实现时我习惯返回一个 JSON 结构把 tasks_base_url 暴露出来GET /.well-known/agent-client-protocol { protocol_version: 0.1, endpoints: { tasks_base_url: https://agent.example.com/acp } }鉴权方面最常见的做法是 HTTP Bearer Token。客户端在请求头里放 Authorization 头服务端校验后把会话信息绑定到任务上下文。我自己测试时用的是方便调试的静态 Token但生产环境我强烈建议接入标准的 OAuth2 流程至少要做到“按用户隔离任务列表”否则一个同学创建的私密任务另一个同学可能直接在客户端看到了。会话标识也要尽早设计。不是每个任务都需要长期会话但如果你的场景是“同一个业务请求可以连续追问”那么每个任务挂一个 session_id后续提交消息时带上这个 ID服务端才能把上下文串联起来。没有会话标识的协议就像聊天工具不带历史记录每轮都是失忆的对话。3. 实操三分钟跑通一个符合 ACP 的服务端原型3.1 准备环境与目录结构纸上谈兵到此为止下面进入实操。我本地用的是 Python 3.11 FastAPI主要考虑到生态成熟、起服务快适合做协议原型。目录结构我建议这样组织acp-demo/ ├── server.py ├── client.py ├── requirements.txt └── README.mdrequirements.txt 里只放三个依赖fastapi、uvicorn、sse-starlette。sse-starlette 是专门用来处理 Server-Sent Events 的扩展库比手动写流式响应省事得多。装好之后打开终端就要开始搭最核心的几个端点。3.2 用 FastAPI 实现核心端点我把服务端拆成三个核心端点创建任务、提交交互消息、获取任务详情。为了演示权限流程我再加一个请求权限的异步处理逻辑。先看创建任务的简化版实现# server.py import uuid import asyncio from datetime import datetime, timezone from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel from sse_starletteseye.sse import EventSourceResponse app FastAPI() tasks_store {} class TaskCreateRequest(BaseModel): title: str description: str metadata: dict {} app.post(/acp/tasks/create) async def create_task( body: TaskCreateRequest, authorization: str Header(default) ): if not authorization.startswith(Bearer ): raise HTTPException(status_code401, detail需要 Bearer Token) task_id ftask_{uuid.uuid4().hex[:12]} now datetime.now(timezone.utc).isoformat() tasks_store[task_id] { id: task_id, status: pending, messages: [ {role: user, content: body.description} ], interactive_turns: [], outputs: [], created_at: now, updated_at: now, } # 模拟一个异步执行流程立刻进入 working asyncio.create_task(_run_task(task_id)) return { id: task_id, status: pending, interactive_turns: [], created_at: now, }这里有一个我后来才补上的关键点创建任务接口一定不要同步等待整个任务执行完。如果任务需要几十秒甚至几分钟客户端请求早就超时了。正确做法是立即返回任务 ID让执行逻辑在后台跑。上面的_run_task就是一个后台协程它负责推进任务状态、模拟工具调用、必要时发起权限请求。再看提交交互消息和事件流推送。事件流是 ACP 适合做实时界面的核心我直接用一个 SSE 端点把任务状态变化推给客户端app.post(/acp/tasks/{task_id}/submit_message) async def submit_message(task_id: str, body: dict): if task_id not in tasks_store: raise HTTPException(status_code404, detail任务不存在) task tasks_store[task_id] task[messages].append({role: user, content: body.get(content, )}) task[status] working task[updated_at] datetime.now(timezone.utc).isoformat() # 触发后台继续干活 asyncio.create_task(_continue_run(task_id)) return {task_id: task_id, status: task[status]} app.get(/acp/tasks/{task_id}/events) async def task_events(task_id: str): SSE 事件流把任务状态变化实时推给客户端 async def event_generator(): while True: task tasks_store.get(task_id) if not task: break event { task_id: task_id, status: task[status], interactive_turns: task[interactive_turns], } # 每 0.5 秒推一次真实实现中可以只在状态变化时推送 yield {event: task_update, data: json.dumps(event, ensure_asciiFalse)} await asyncio.sleep(0.5) return EventSourceResponse(event_generator())看到这里你应该发现SSE 本质上就是服务端单向推送给客户端的通道客户端用 EventSource 或者 fetch 流式读取即可。这里注意一点我演示的循环推送是为了简化问题生产环境里直接在状态变化时 push 一次可以配合 asyncio.Queue 做事件总线。3.3 客户端接入示例与事件消费服务端搭好后客户端接入就很简单。用 Python 的 httpx 发起创建任务请求用 sseclient 消费事件流。下面这段代码是我本地验证时实际跑过的curl -X POST http://localhost:8000/acp/tasks/create \ -H Authorization: Bearer test-token \ -H Content-Type: application/json \ -d {title: 测试任务,description: 把当前目录的文件清单列出来}返回里拿到任务 ID 后开另一个终端窗口消费事件# client.py import httpx import json resp httpx.post( http://localhost:8000/acp/tasks/create, headers{Authorization: Bearer test-token}, json{title: 测试任务, description: 列出文件清单}, ) task resp.json() print(task created:, task[id]) # 消费事件流 with httpx.stream( GET, fhttp://localhost:8000/acp/tasks/{task[id]}/events, headers{Authorization: Bearer test-token}, timeout30, ) as r: for line in r.iter_lines(): if line.startswith(data: ): payload json.loads(line[6:]) print(payload[status], payload[interactive_turns])这个示例是我在原型阶段最常用的骨架。你可以把任务提交、事件推送、权限审批三条链路同时跑起来看到的效果是客户端创建任务后服务端把状态从 pending 推到 working然后推进到 completed整个过程没有任何轮询代码。3.4 本地验证与压测的心得我跑这个原型的过程中有一个体会事件流别只测试“正常流程”一定要专门测试服务端把任务卡在 needs_input 的场景。比如权限请求发出后客户端迟迟不响应服务端能不能在超时后准确地把任务标记为 failed我在本地用 3 秒超时测了一遍发现如果没有超时保护任务会一直挂着前端界面永远停在“请求审批中”。这个问题不是协议设计复杂而是实现里很容易忽略。另外压测时我踩过一个具体问题事件流接口被高并发访问后会创建大量异步任务对象内存涨得很快。后来我加上简单的事件总线把 SSE 消费端连接到共享队列而不是每次请求都重复拉取任务状态内存占用降了一个数量级。这个小优化我推荐你在原型阶段就做进去不然后期改起来工作量大。4. 协议落地中的关键决策与避坑指南4.1 长任务一定要用异步任务模型这是 ACP 服务端最容易犯的错误。很多第一次接触协议的同学会把它当成普通 REST API在创建任务接口里同步执行完所有步骤最后返回最终结果。这样做在小 demo 里没问题一旦任务超过 HTTP 默认超时时间客户端立刻收到连接中断任务状态还卡在 pending。我的经验是创建任务和提交消息都只负责“接收意图”真正的执行放在后台队列或者消息队列里客户端通过事件流感知进度。异步模型带来的好处不只是不超时还让系统多了暂停和恢复能力。服务端可以在等待用户审批时挂起整个任务用户响应后重新唤醒这在同步模型里几乎不可能优雅实现。4.2 权限审批必须能“暂停”和“唤醒”权限审批是 ACP 里最考验实现的环节。我自己的实现方式是给每个任务挂一个 asyncio.Eventpermission_event asyncio.Event() permission_result None # 服务端请求权限 async def require_permission(task_id, permission_request): global permission_result permission_result None # 推送权限请求给客户端 await send_permission_request(task_id, permission_request) # 阻塞等待客户端响应 try: await asyncio.wait_for(permission_event.wait(), timeout300) except asyncio.TimeoutError: return denied result permission_result permission_event.clear() return result # 客户端响应后唤醒 def respond_permission(task_id, request_id, decision): global permission_result permission_result decision permission_event.set()这种“阻塞-唤醒”模型在异步编程里是标配写起来简单调试时也很直观。更重要的是它让服务端不需要维护复杂的轮询循环。权限请求推给客户端那一刻服务端只知道“在等人”具体等多久由超时时间控制。4.3 超时、重试与幂等设计协议落地时超时与重试是隐藏的魔鬼。我踩过最典型的一个坑是客户端重试创建任务时因为网络抖动同一个任务被创建了两遍产生了两个任务 ID后面所有追踪全部错乱。解决办法是给创建任务请求加一个幂等键Idempotency-Key服务端先查幂等键如果已经存在就直接返回旧任务不再重复创建。重试策略上我以 500 和 503 错误为例遇到 500 先别急着冲进去改代码可能是任务本身执行失败遇到 503 通常是服务端过载适合做指数退避重试。但重试之前一定要确认请求是幂等的否则同样的工具调用会被执行多次邮件发两遍、订单建两次后果很严重。超时方面要分层次设置HTTP 请求超时、SSE 事件流空闲超时、权限请求超时、整体任务超时。每一层都单独配置不要一刀切。权限类的超时我一般给 5 分钟普通任务超时给 30 分钟后台工具调用超时给 30 秒各有各的合理性。4.4 多租户与配额管理如果 ACP 服务端要服务多个业务方必须尽早考虑租户隔离。我在实际设计里每一个任务都绑定租户 ID然后所有接口都检查“当前鉴权用户是否有权访问该任务”。我不会把租户隔离做成事后加补丁——那意味着整个消息结构都要改成本翻倍。配额管理也建议放在协议层任务创建前检查当前租户剩余配额并发任务数超限直接返回 429 或者明确错误码。事件流连接是配额消耗的大头每个 SSE 连接都占一个文件描述符连接数上限要单独设置不然某个租户异常重连就可能把整个服务端资源拖垮。4.5 事件流连接与反向代理的兼容性事件流在本地跑得好好的一上到反向代理层就出现“事件不更新”的怪现象。这个坑我印象很深大部分反向代理默认会缓冲响应内容SSE 事件被缓冲后客户端等半天也收不到新消息。解决办法非常明确给事件流响应显式加上X-Accel-Buffering: no头同时确保Content-Type: text/event-stream没有被压缩或者篡改。如果用了多云负载均衡还要确认没有额外的代理层做同样的缓冲不然每一个代理层都可能让事件流“断流”。5. 常见问题与排查技巧实录5.1 排查问题速查表我在做模拟项目 X 和后面迭代 ACP 服务端的过程中整理了一份问题排查表今天一次性放出来现象可能原因排查/解决办法创建任务后状态一直是 pending后台协程没被触发或执行异常检查服务端日志确认_run_task被调用并捕获异常客户端收不到 SSE 事件反向代理缓冲了响应设置X-Accel-Buffering: no确认响应头正确权限请求发出后客户端无感知事件流没有推送权限事件检查权限请求是否走独立 SSE 事件类型客户端是否订阅任务重复创建客户端重试导致重复提交引入幂等键服务端做去重任务执行到一半失败工具调用超时或权限被拒查看任务事件流里的错误事件检查工具超时配置消息内容乱码编码/压缩问题统一使用 UTF-8关闭事件流压缩任务状态错乱并发修改了同一个任务对象服务端加锁或用队列保证单任务串行处理高并发时内存暴涨每个 SSE 连接重复扫描任务列表引入事件总线按任务 ID 分发事件这张表我建议团队内部直接贴在 wiki 里。很多问题不是协议本身有坑而是实现细节里没有对齐一张速查表能省掉大量重复排查。5.2 我遇到过的最隐蔽的一个坑最隐蔽的坑是“SSE 连接被错误复用”。客户端用了连接池后同一个 HTTP/1.1 连接可能被多个请求复用而一个 SSE 事件流是长连接一旦被连接池错误复用后面的普通请求就会被“挂”在事件流上永远没有响应。这个问题的症状非常诡异部分请求偶尔卡死重启客户端又恢复正常。排查办法是在服务端查看连接生命周期事件流接口必须单独配置连接策略关闭连接复用。我在客户端里对/events端点强制使用新连接问题才消失。这种问题常规日志不太容易发现如果你也遇到“模模糊糊的偶发卡死”可以往这个方向查一查。5.3 从协议走向真实产品的小建议如果你的目标不是写 demo而是把 ACP 用进真实产品我的建议是先做内部规范再逐步对外开放。第一步把任务状态、消息结构、权限审批三部分按本文的骨架冻结成文档第二步让客户端和服务端并行开发以事件流为唯一权威状态源第三步引入契约测试确保消息结构变更不会破坏已发布客户端。我个人在实际操作中体会最深的一点是不要一开始追求完整实现协议里所有消息而是先把“创建任务—事件推送—权限审批—结果回传”这条主链路跑通。主链路稳定之后其他消息都是增量扩展。ACP 的价值就在于它给了你一个稳定的通信骨架剩下的业务复杂度可以安心地往里填充而不需要一次次推翻接口设计。最后再分享一个小技巧把 ACP 的事件流内容全部结构化打印到本地日志里前端界面出了任何状态问题先查日志再查代码。事件流是协议的真实执行轨迹它能帮你还原整个任务从开始到结束的完整过程省下来的排查时间非常可观。