不写 Agent Loop：用 ACP 快速搭一个能干活的 Agent

为了做一个 Agent，先写半个 Agent 框架，这事多少有点本末倒置。

之前做飞书助手时，我一开始想的是常规路线：接模型 API、定义 Function Calling、实现 Agent Loop，再把文件、Shell 和飞书能力一个个封装成工具。列完待办之后才发现，真正的业务代码还没写，Agent 运行时倒是快搭出一套了。

但我电脑里本来就有 Cursor Agent。它已经会调模型、读写文件、跑命令，也能使用 Rules、Skills 和 MCP。缺的只是一种办法，让我的 Python 程序像编辑器一样去驱动它。

ACP 正好补上了这一层。这篇不展开讲一长串协议史，只做一件事：用 ACP 把现成的 Agent 能力接进自己的程序，先跑起来，再看怎么扩成一个真正能干活的 Agent。

在 ACP 之前，我们是怎么接 AI 的

在写 ACP Client 之前，先花一点篇幅把前面的几块拼图摆好。Chat Completions、Responses API、MCP 和 ACP 经常一起出现，但它们解决的并不是同一个问题。

Chat Completions：模型只负责回答

最直接的接法是 Chat Completions API：程序组装 messages，模型返回文字；需要多轮对话，就由程序保存历史并在下一次请求时继续带上。

模型后来也能返回 tool_calls，告诉程序「我想调用哪个函数、参数是什么」。但模型只负责提出调用意图，真正执行函数、回填结果和决定是否继续请求的，仍然是我们的程序。

这里特意写 Chat Completions，是因为它和更早的 Completions API 不是同一个接口：前者接收 messages，后者主要接收 prompt。很多文章把两者都简称为 Completion，读代码时很容易串台。

Responses API：少写一些状态和工具胶水

Responses API 把文本、图片、工具调用等结果放进统一的 response 结构里，也提供 Web Search、File Search、Code Interpreter 等内置工具。多轮对话可以用 previous_response_id 串起来，不必每次都重新传一份完整的 messages。

不过，这不等于模型有了无限记忆。关联的历史仍会占用上下文和 token，也仍然受上下文窗口限制。使用自定义函数时，业务程序依旧要执行工具并把结果交回模型，Agent Loop 并没有完全消失。

MCP：把 Agent 接工具的方式统一起来

当每个客户端都要单独接文件、数据库和内部 API 时，工具集成很快会变成另一堆胶水。MCP（Model Context Protocol）给这层定义了统一接口，让 MCP Server 可以暴露 tools、resources 和 prompts，支持 MCP 的 Host 再通过 MCP Client 使用这些能力。

MCP 解决的是「Agent 怎么接外部能力」，不负责替业务程序驱动一个完整 Agent。任务怎么规划、工具何时调用、失败后是否重试，仍然需要 Agent 运行时或自己的编排代码处理。

走到这里，模型能回答，工具也能标准化接入，但我的 Python 程序如果想直接使用一个完整 Agent，仍然缺少 Client 和 Agent 之间的接口。ACP 补的就是这一层。

自己接模型，麻烦主要不在那次 HTTP 请求

只做一次问答，调用模型 API 很简单。等模型开始调工具，事情就变成了一个循环：

这个循环就是 Agent Loop。真正费时间的地方都藏在循环周围：

• 保存和裁剪对话上下文；
• 注册工具、校验参数、回填结果；
• 处理重试、超时和中途取消；
• 决定敏感工具要不要人工批准；
• 把模型的流式输出和工具状态交给上层界面。

这些工作并非没有价值。金融交易、发布上线、数据删除之类的流程，需要确定性和完整审计，自己掌握 Agent Loop 更稳。

但我当时想做的是个人工具和飞书助手，核心需求是让 Agent 读项目、查资料、跑脚本。为了这些能力重新造一遍运行时，投入有点重。ACP 的思路是把分工换一下：业务程序只当 Client，规划和工具循环交给已经成熟的 Agent。

ACP 到底接走了哪部分

ACP 全称 Agent Client Protocol，约定了 Client 和 Agent 怎么通信。这里的 Client 可以是编辑器，也可以是 Python 服务、聊天机器人或桌面工具；Agent 则是 Cursor 这类已经具备模型和工具能力的运行时。

我更愿意把这层关系理解成一个木偶师舞台：图左上角的控制台是自己的业务程序，中间三个发光核心代表不同的 Coding Agent，底部机械结构则是各家 Agent 运行时长期积累的 Harness。自己的程序只牵住 ACP 这根控制杆，用统一方式给 Coding Agent 下任务、维持会话、处理权限并接收结果；具体怎么规划和调用能力，仍由 Coding Agent 自己决定。

图里的 Harness 不是 ACP 协议规定的某个模块，而是业界 Coding Agent 背后那套长期打磨的工程底座：上下文管理、任务规划、工具循环、权限确认、会话恢复、错误处理和可观测性。它们才是一个 Agent 从「能调模型」走到「能稳定干活」最费功夫的部分。想继续拆这层幕后工程，可以看 Harness Engineering 的介绍与编程实践。

所以 ACP 带来的不只是少写一次 API 调用。我的程序通过同一套 Client 接口驱动成熟 Coding Agent，Agent 再去使用自己的 Tools、Skills 和 MCP，相当于把这些产品已经积累的 Harness Engineering 一起复用了。木偶师只表达意图和边界，并不微操木偶的每一个关节。

接入成熟 Coding Agent 后，我自己的程序仍要负责业务入口、身份、消息路由和安全策略，模型调用、任务规划、工具循环和内部状态则由 Agent 运行时承担。ACP 做的事情更克制：把 Client 与 Agent 之间的交互接口统一起来。

这和普通的 headless CLI 也有区别。很多 AI CLI 都能接收 prompt、输出 JSON，但参数、事件类型和会话恢复方式各有一套。ACP 把初始化、能力协商、session、流式更新、权限请求和取消这些生命周期统一成了协议。

要注意，能从 stdin 读写 JSON 不代表支持 ACP。具体工具能不能接，要看它是否明确实现了 ACP。

先跑通最小链路

下面用 Cursor CLI 做 Agent。先安装并登录，然后确认 ACP 模式能启动：

agent login
agent acp --help

agent acp 启动后不会出现聊天界面。它在 stdin 等待 JSON-RPC 请求，再把 response 和 notification 一行一条写到 stdout。

一次完整提问要经过这几步：

这几个方法里，initialize 建立协议能力，session/new 创建对话，session/prompt 才真正把任务交给 Agent。Agent 的文字会通过 session/update 持续返回，收到对应的 session/prompt response 才表示这一轮结束。

一个只依赖标准库的 ACP Client

下面这个例子尽量压到最小，只做一轮只读问答。它会把 session 切到 Cursor 的 ask mode，不执行写文件或 Shell 等工具。代码虽然不长，但走的是真实 ACP 流程，不是把几个自定义的 message、text、done 事件拼成示意代码。

还有一个边界要先说：这个版本是单进程、单线程、一次只跑一个 turn 的教学示例。它适合验证协议链路，不适合直接拿去同时处理多个群聊。

import json
import subprocess
from pathlib import Path


class CursorAgent:
    def __init__(self, workdir="."):
        self.workdir = Path(workdir).resolve()
        self.proc = subprocess.Popen(
            ["agent", "acp"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=self.workdir,
            text=True,
            bufsize=1,
        )
        self.next_id = 1
        self.chunks = []
        self.active_session_id = None

    def send(self, message):
        self.proc.stdin.write(json.dumps(message) + "\n")
        self.proc.stdin.flush()

    def request(self, method, params):
        request_id = self.next_id
        self.next_id += 1
        message = {
            "jsonrpc": "2.0",
            "id": request_id,
            "method": method,
            "params": params,
        }
        self.send(message)

        for line in self.proc.stdout:
            incoming = json.loads(line)

            if incoming.get("method") == "session/update":
                params = incoming.get("params", {})
                update = params.get("update", {})
                if (
                    params.get("sessionId") == self.active_session_id
                    and update.get("sessionUpdate") == "agent_message_chunk"
                ):
                    text = update.get("content", {}).get("text", "")
                    self.chunks.append(text)
                continue

            # 这是只读教学示例，不支持权限、提问或 Plan 等反向请求。
            # 返回明确错误，避免 Agent 和 Client 互相等待。
            if "method" in incoming and "id" in incoming:
                self.send(
                    {
                        "jsonrpc": "2.0",
                        "id": incoming["id"],
                        "error": {
                            "code": -32601,
                            "message": "Client method not implemented in read-only demo",
                        },
                    }
                )
                continue

            if incoming.get("id") != request_id:
                continue
            if "error" in incoming:
                raise RuntimeError(incoming["error"])
            return incoming["result"]

        raise RuntimeError("Cursor Agent 意外退出")

    def start(self):
        self.request(
            "initialize",
            {
                "protocolVersion": 1,
                "clientCapabilities": {
                    "fs": {"readTextFile": False, "writeTextFile": False},
                    "terminal": False,
                },
                "clientInfo": {"name": "python-acp-demo", "version": "0.1.0"},
            },
        )
        self.request("authenticate", {"methodId": "cursor_login"})
        session = self.request(
            "session/new",
            {"cwd": str(self.workdir), "mcpServers": []},
        )
        session_id = session["sessionId"]

        modes = session.get("modes", {})
        available_modes = {mode["id"] for mode in modes.get("availableModes", [])}
        if "ask" not in available_modes:
            raise RuntimeError("当前 Cursor Agent 不支持 ask mode")
        if modes.get("currentModeId") != "ask":
            self.request(
                "session/set_mode",
                {"sessionId": session_id, "modeId": "ask"},
            )

        return session_id

    def ask(self, session_id, prompt):
        self.chunks.clear()
        self.active_session_id = session_id
        self.request(
            "session/prompt",
            {
                "sessionId": session_id,
                "prompt": [{"type": "text", "text": prompt}],
            },
        )
        return "".join(self.chunks)

    def close(self):
        if self.proc.poll() is not None:
            return
        self.proc.stdin.close()
        self.proc.terminate()
        try:
            self.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.proc.kill()
            self.proc.wait()


if __name__ == "__main__":
    agent = CursorAgent()
    try:
        session_id = agent.start()
        print(agent.ask(session_id, "用一句话解释 ACP 是什么"))
    finally:
        agent.close()

这段代码做完了三件关键的事：

1. 把 Cursor Agent 作为子进程启动；
2. 用 JSON-RPC 建立并保留一个 session；
3. 从 session/update 中拼出 Agent 的流式回复。

Client 没有导入模型 SDK，也没有配置 OPENAI_API_KEY。模型认证和 Agent 工具由 Cursor 托管，但这不代表模型免费或离线运行，只是账号和调用入口不再落到这段业务代码里。

代码里还有两处刻意的收口。cwd 会先解析成绝对路径，这是 ACP 对 session/new 的要求；遇到示例没有实现的 Agent → Client 反向请求时，它会返回 JSON-RPC Method not found，避免双方一直等下去。真正要让 Agent 调工具，不能靠这条兜底。

从「能聊天」到「能干活」还差什么

上面的最小例子只证明链路通了。一个能长期运行的 Agent，还得补几个工程问题。

工作目录决定 Agent 能看到什么

session/new 的 cwd 不只是普通参数。它决定 Agent 从哪里读取项目文件、Rules 或 AGENTS.md，也决定相对路径落在哪里。

示例已经把工作目录解析成绝对路径。正式代码还要把它限制在明确的项目范围内，不要让一个来自聊天软件的 prompt 默认拿到整个用户目录。

Session 要跟业务身份绑定

最小示例每次启动都创建一个 session。接到业务系统后，通常要建立一层映射：

飞书 chat_id  → ACP session_id
网页 user_id  → ACP session_id
工单 ticket_id → ACP session_id

这样同一个群或同一张工单里的多轮消息才能沿用上下文，不同用户也不会串话。Agent 支持 session/load 时，还可以在进程重启后恢复会话。

这里不能只把示例里的 chunks 改成字典就完事。一个 Agent 子进程只有一条 stdout，多个线程同时读取会拿走彼此的 response，session/update 也可能串到另一个会话。正式 Client 应该只有一个 reader，再用 request ID 把 response 分发给 pending request，用 sessionId 把更新送到各自缓冲区；同一个 session 的 turn 还要串行执行。

权限请求不能假装不存在

Agent 要读文件、执行 Shell 或调用 MCP 时，可能向 Client 发起 session/request_permission。Cursor 还可能发出 cursor/ask_question、cursor/create_plan 等扩展请求。桌面编辑器可以弹窗，无人值守的机器人却没人点按钮。

这类消息是 Agent → Client 的 JSON-RPC request，必须由统一 reader 交给对应 handler 并回传 response。教学示例切到了 ask mode，而且对未知反向请求返回错误；工具型 Agent 则要按产品能力实现权限、提问和 Plan handler，不能直接忽略消息，也不能看到权限请求就一律批准。

最省事的做法是自动批准，但这等于让外部消息具备了触发本机工具的能力。我的处理思路是：

• 只允许白名单用户或群聊进入 ACP；
• 只开放专用工作目录；
• 查询类动作可以自动批准；
• 删除、发布、发消息等写操作增加业务确认；
• 确定性的运维动作直接走固定脚本，不交给模型自由发挥。

权限这一层没想清楚，Agent 越能干，风险越大。

进程和事件流也要当服务维护

教学示例把 stderr 指向了 DEVNULL，避免无人读取时把管道塞满。正式使用则应该持续消费并保存日志；每个请求要有超时；Agent 异常退出后要清理 session 并决定是否重启。收到取消信号时，也应该通过 session/cancel 结束当前 turn。若 Agent 又拉起了其他子进程，还要按操作系统处理整个进程组，而不只是杀掉最外层进程。

关于逐行处理 JSON 消息，可以参考站内的 NDJSON：Agent 开发离不开的流式数据格式。

接上业务入口，一个 Agent 就成形了

ACP Client 本身不关心消息来自哪里。最小链路跑通后，外面包一层业务适配器就够了：

业务入口 → 身份与权限校验 → prompt 组装 → ACP session → Agent 回复 → 原路返回

入口可以是飞书消息、HTTP API、桌面窗口、定时任务，也可以是另一个自动化程序。业务层负责「谁在什么场景下提出了什么请求」，Agent 负责「为了完成请求该调用哪些能力」。

我在站内的 利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口 就是这个最小结构的完整案例：

• lark-cli 负责收发飞书消息；
• Python 负责过滤、路由和按 chat_id 复用 session；
• Cursor ACP 负责开放问题的规划与工具调用；
• AGENTS.md 约束 Agent 的角色和操作边界；
• 固定回复与部署脚本绕过模型，走确定性流程。

那篇文章更偏「一个飞书 Agent 怎样落地」，这篇解决的是它前面的一步：为什么可以不从模型 API 开始，以及 ACP Client 最小要写哪些东西。

哪些场景适合这样做

ACP 很适合：

• 已经在使用某个支持 ACP 的 Agent，希望复用它的模型、工具和配置；
• 快速验证个人助手、内部机器人或开发工具；
• 业务层只想处理入口和权限，不想维护 Agent Loop；
• 希望以后替换 Agent 时，Client 不跟某个模型 SDK 深度绑定，同时愿意处理不同 Agent 的可选能力和扩展差异。

下面这些情况，我会更倾向自己掌握模型调用和编排：

• 每一步工具调用都要做严格审计；
• 延迟、token 和模型选择需要精细控制；
• 流程高度确定，根本不需要 Agent 自由规划；
• 运行环境不能依赖本地 Agent 进程或它的登录态。

ACP 省掉的是重复搭 Agent 运行时的工作，并没有消除工程成本。只是你可以把精力花在业务入口、权限边界和真正有差异的能力上，而不是再次实现工具循环。

结语

用 ACP 搭 Agent，最吸引我的地方不是协议本身有多漂亮，而是它把「我想做一个 Agent」缩短成了「我给现成 Agent 接一个入口」。

先用几十行代码把链路跑通，再补 session、权限和业务适配器，一个能干活的 Agent 就有了轮廓。至于那些确定性很强的事情，还是老老实实写脚本——能一行 Bash 说清楚的，没必要让模型猜。

参考资料

• Agent Client Protocol：Overview
• Agent Client Protocol：Prompt Turn
• Agent Client Protocol：Clients
• Cursor CLI：ACP
• OpenAI：Conversation state
• OpenAI：Using tools
• Model Context Protocol：Architecture overview

版权声明: 本文首发于 指尖魔法屋-不写 Agent Loop：用 ACP 快速搭一个能干活的 Agent（https://blog.thinkmoon.cn/post/1035-build-agent-with-acp/） 转载或引用必须申明原指尖魔法屋来源及源地址！