企业官网建设流程全解析

在智能体应用还停留在“谁的客户端就配谁的私有接口”那个阶段时，我第一次看到 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 与“模型上下文协议”的定位差异

说到这里，很多人会自然想到现在社区里热门的 MCP（Model Context Protocol）这类面向模型和工具之间上下文交换的协议。它们之间不是替代关系，更多是互补。

可以把 MCP 理解成“模型外挂工具的标准插座”。模型要查数据、要读写文件、要调用业务系统，都是通过 MCP 去发现工具、传递上下文、拿回工具结果。它解决的是模型这一层如何连接外部能力的问题。而 ACP 解决的是“一个应用客户端如何把一个任务委托给智能体服务，并持续跟踪处理过程”的问题。它更偏业务协作流：任务谁发起、状态谁维护、权限谁审批、结果谁接收。

在实际系统里，两者可以同时存在。我去年做的一个模拟项目 X，客户端是一个对话工作台，它通过 ACP 把“帮我准备月度报表”这个任务提交给智能体服务端；服务端内部再用 MCP 调数据库工具、文件聚合工具、导出工具。前端完全不需要知道后端到底调用了哪些工具，这就是分层带来的价值。

1.3 为什么不能“直接用 WebSocket 随便做”

一定要争辩的话，WebSocket 当然可以做类似的通信。但关键问题是：WebSocket 只给了一个全双工通道，它不管消息格式、错误码、权限流程、任务生命周期这些东西。你还是得自己定义一套提交流程、一套状态机、一套重试约定。ACP 相当于把这些已经踩过无数遍坑的设计整理成了现成的规范。你可以把它理解成“智能体世界的 HTTP”——HTTP 本身也不实现业务逻辑，但它定义了方法、状态码、头部这些基础知识，所有人都按这个来，生态才能互通。

所以我的建议很明确：如果只是自己内部联调，随便怎么玩都行；如果想让前端应用能够对接多个智能体服务、甚至未来开源出去让别人接入，早点拥抱 ACP 这种标准协议，比后期再做适配成本低得多。

2. Agent Client Protocol 的核心消息结构与状态机

2.1 任务生命周期与五类核心消息

ACP 的核心对象是“任务”。一个任务从创建到结束，会经过几个稳定的状态，我用下面的表来整理：

围绕任务的生命周期，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": "ops@example.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.md

requirements.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_code=401, detail="需要 Bearer Token") task_id = f"task_{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_code=404, 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_ascii=False)} 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", f"http://localhost:8000/acp/tasks/{task['id']}/events", headers={"Authorization": "Bearer test-token"}, timeout=30, ) 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.Event：

permission_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(), timeout=300) 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 服务端的过程中，整理了一份问题排查表，今天一次性放出来：

这张表我建议团队内部直接贴在 wiki 里。很多问题不是协议本身有坑，而是实现细节里没有对齐，一张速查表能省掉大量重复排查。

5.2 我遇到过的最隐蔽的一个坑

最隐蔽的坑是“SSE 连接被错误复用”。客户端用了连接池后，同一个 HTTP/1.1 连接可能被多个请求复用，而一个 SSE 事件流是长连接，一旦被连接池错误复用，后面的普通请求就会被“挂”在事件流上，永远没有响应。这个问题的症状非常诡异：部分请求偶尔卡死，重启客户端又恢复正常。

排查办法是在服务端查看连接生命周期，事件流接口必须单独配置连接策略，关闭连接复用。我在客户端里对/events端点强制使用新连接，问题才消失。这种问题常规日志不太容易发现，如果你也遇到“模模糊糊的偶发卡死”，可以往这个方向查一查。

5.3 从协议走向真实产品的小建议

如果你的目标不是写 demo，而是把 ACP 用进真实产品，我的建议是先做内部规范，再逐步对外开放。第一步，把任务状态、消息结构、权限审批三部分按本文的骨架冻结成文档；第二步，让客户端和服务端并行开发，以事件流为唯一权威状态源；第三步，引入契约测试，确保消息结构变更不会破坏已发布客户端。

我个人在实际操作中体会最深的一点是：不要一开始追求完整实现协议里所有消息，而是先把“创建任务—事件推送—权限审批—结果回传”这条主链路跑通。主链路稳定之后，其他消息都是增量扩展。ACP 的价值就在于它给了你一个稳定的通信骨架，剩下的业务复杂度可以安心地往里填充，而不需要一次次推翻接口设计。

最后再分享一个小技巧：把 ACP 的事件流内容全部结构化打印到本地日志里，前端界面出了任何状态问题，先查日志再查代码。事件流是协议的真实执行轨迹，它能帮你还原整个任务从开始到结束的完整过程，省下来的排查时间非常可观。