Agent 运行时: 从消息循环到 LLM 调用

用红绿灯建立直觉

后面会反复出现"状态机""消息循环"两个词——它们是 Agent 内部的两层抽象,先用一个熟悉的例子建立直觉。

状态机:静止的规则图

   ┌─────┐  30s    ┌─────┐  5s    ┌─────┐
   │ 绿  │ ──────> │  黄  │ ──────> │  红  │
   └─────┘        └─────┘        └─────┘
     ↑                                │
     │                                │
     └────────── 30s ─────────────────┘
                 (回到绿,循环)

• 方框 = 状态(绿 / 黄 / 红)
• 箭头 = 可以怎么转(单向,只有这一条路)
• 箭头上的字 = 触发条件(30s / 5s / 30s)

这张图本身是静止的——你不动它,十字路口永远是绿,谁都不变。

消息循环:不停执行规则的人

让红绿灯真转起来,需要一段"不停在做事的程序":

// 状态机(规则):一张表
const TRANSITIONS = {
  绿: { next: "黄", after: 30 },
  黄: { next: "红", after: 5 },
  红: { next: "绿", after: 30 },
};

// 消息循环(执行):不停转
let state = "绿";
while (true) {
  sleep(TRANSITIONS[state].after * 1000);  // 阻塞等时间到
  state = TRANSITIONS[state].next;          // 改状态(状态机被动地被改)
  lightOn(state);                            // 副作用:亮灯
}

两者的关系:

• 状态机是被动的——它只记录"现在到哪",自己不主动变
• 消息循环是主动的——它不停等事件(时间到就是事件)、读状态、按规定改状态、触发副作用
• 状态机永远静止;消息循环永远在转
• 没有状态机,消息循环不知道现在该做啥
• 没有消息循环,状态机永远是绿

s01_agent_loop

https://learn.shareai.run/zh/s01/

import os
import subprocess
from pathlib import Path
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent / ".env", override=True)

client = Anthropic(
    base_url=os.getenv("ANTHROPIC_BASE_URL"),
    auth_token=os.getenv("ANTHROPIC_AUTH_TOKEN"),
)
MODEL = os.environ["MODEL_ID"]

# 系统提示词
SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

# 工具声明列表，把“可用工具”描述给模型，这个结构不是随便自定义的，它是 Anthropic Claude Messages API 的 Tool Use / Function Calling 工具定义格式
# | 字段 | 作用 |
# |---|---|
# | `name` | 工具名，模型调用工具时会用这个名字 |
# | `description` | 工具说明，帮助模型判断什么时候用这个工具 |
# | `input_schema` | 工具入参格式，告诉模型调用工具时应该传什么参数 |

# Anthropic 官方文档也说明：如果要让 Claude 调用你自己定义的函数，需要传一个带 input_schema 的 tool；Claude 会返回 tool_use，然后由你的应用程序执行这个调用。可参考：https://platform.claude.com/docs/zh-CN/agents-and-tools/tool-use/overview
TOOLS = [
    {
        "name": "bash",
        "description": "Run a shell command.",
        "input_schema": {
            "type": "object",  # 表示工具输入必须是一个对象。
            "properties": {
                "command": {"type": "string"}
            },  # 表示对象里允许/定义了一个字段 command，类型是字符串。
            "required": ["command"],  # 表示 command 是必填参数。
        },
    }
]

# 所以模型调用上面这个工具时，应该生成类似：
# {
#   "type": "tool_use",
#   "name": "bash",
#   "input": {
#     "command": "pwd"
#   }
# }


def run_bash(command: str) -> str:
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"
    try:
        r = subprocess.run(
            command,
            shell=True,
            cwd=os.getcwd(),
            capture_output=True,
            text=True,
            timeout=120,
        )
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"
    except (FileNotFoundError, OSError) as e:
        return f"Error: {e}"


# -- The core pattern: a while loop that calls tools until the model stops --
def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000,
        )
        # Append assistant turn
        messages.append({"role": "assistant", "content": response.content})
        # If the model didn't call a tool, we're done
        if response.stop_reason != "tool_use":
            return
        # 模型如果想执行命令，会生成类似这样的 tool call：
        # {
        #     "type": "tool_use",
        #     "name": "bash",
        #     "input": {
        #         "command": "ls -la"
        #     }
        # }
        # Execute each tool call, collect results
        results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"\033[33m$ {block.input['command']}\033[0m")
                # 当前这里为了演示，写死一个调用工具，如果有多个工具，更通用的写法：用工具分发表
                output = run_bash(block.input["command"])
                print(output[:200])
                results.append(
                    {"type": "tool_result", "tool_use_id": block.id, "content": output}
                )
        messages.append({"role": "user", "content": results})


if __name__ == "__main__":
    history = []
    while True:
        try:
            query = input("\033[36ms01 >> \033[0m")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        history.append({"role": "user", "content": query})
        agent_loop(history)
        response_content = history[-1]["content"]
        if isinstance(response_content, list):
            for block in response_content:
                if hasattr(block, "text"):
                    print(block.text)
        print()

1. input 是工具调用结果里的固定字段名

当模型决定调用工具时，Anthropic API 返回的 tool_use 块大概会长这样：

{
  "type": "tool_use",
  "id": "toolu_xxx",
  "name": "bash",
  "input": {
    "command": "pwd"
  }
}

这里：

• type：表示这是一个工具调用块；
• id：这次工具调用的唯一 ID；
• name：要调用哪个工具，比如 "bash"；
• input：传给这个工具的参数对象。

所以 input 这个字段名是 Anthropic tool use 协议里的固定字段。

tool_use.id 是 Anthropic API 在模型返回工具调用时自动生成的唯一标识；你的程序不用生成它，只需要在回传 tool_result 时用 tool_use_id: block.id 原样带回去，用来匹配“工具调用”和“工具执行结果”。

2. input 里面的内容不是固定的，而是由 input_schema 决定

你定义工具时写的是：

"input_schema": {
    "type": "object",
    "properties": {
        "command": {"type": "string"}
    },
    "required": ["command"],
}

这表示：调用 bash 工具时，input 里面必须有一个字符串字段 command。

所以模型会生成：

"input": {
  "command": "pwd"
}

s02_tool_use

import os
import subprocess
from pathlib import Path

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv(Path(__file__).parent / ".env", override=True)

WORKDIR = Path.cwd()
client = Anthropic(
    base_url=os.getenv("ANTHROPIC_BASE_URL"),
    auth_token=os.getenv("ANTHROPIC_AUTH_TOKEN"),
)
MODEL = os.environ["MODEL_ID"]

SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks. Act, don't explain."


def safe_path(p: str) -> Path:
    # 这里的 `/` 不是数学除法，而是 `pathlib.Path` 里的路径拼接语法。
    # 比如：
    # ```python
    # WORKDIR / "agents/s01_agent_loop.py"
    # ```
    # 等价于：
    # ```text
    # /Users/xxx/learn-claude-code/agents/s01_agent_loop.py
    # ```
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path


def run_bash(command: str) -> str:
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
                           capture_output=True, text=True,
                           encoding="utf-8", errors="replace", timeout=120)
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"


def run_read(path: str, limit: int = None) -> str:
    try:
        text = safe_path(path).read_text()
        lines = text.splitlines()
        if limit and limit < len(lines):
            lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"Error: {e}"


def run_write(path: str, content: str) -> str:
    try:
        fp = safe_path(path)
        fp.parent.mkdir(parents=True, exist_ok=True)
        fp.write_text(content)
        return f"Wrote {len(content)} bytes to {path}"
    except Exception as e:
        return f"Error: {e}"


def run_edit(path: str, old_text: str, new_text: str) -> str:
    try:
        fp = safe_path(path)
        content = fp.read_text()
        if old_text not in content:
            return f"Error: Text not found in {path}"
        fp.write_text(content.replace(old_text, new_text, 1))
        return f"Edited {path}"
    except Exception as e:
        return f"Error: {e}"


# -- The dispatch map: {tool_name: handler} --
TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}

TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
]


def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
                print(f"> {block.name}:")
                print(output[:200])
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})


if __name__ == "__main__":
    history = []
    while True:
        try:
            query = input("\033[36ms02 >> \033[0m")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        history.append({"role": "user", "content": query})
        agent_loop(history)
        response_content = history[-1]["content"]
        if isinstance(response_content, list):
            for block in response_content:
                if hasattr(block, "text"):
                    print(block.text)
        print()

疑问：1

每次循环都要把SYSTEM带上吗？另外，messages岂不是会越来越大？

简单总结：
client.messages.create(...) 是无状态调用，所以每次循环都需要传入 system=SYSTEM，否则模型在该轮请求中就看不到系统提示，行为约束可能失效。tools=TOOLS 也同理，通常也要每次传。

messages 不会指数级增长，而是线性增长：每轮模型回复会追加一条 assistant 消息；如果调用工具，还会追加一条包含 tool_result 的 user 消息。

但要注意，每次请求都会把完整 messages 历史重新发给模型，所以单次请求 token 会越来越多，整个会话累计成本接近 O(n²)。尤其工具输出较大时，历史会膨胀得很快。

实际工程中通常需要做：限制历史长度、压缩旧上下文、裁剪工具输出、按需读取文件片段，避免上下文越来越大。

s03_todo_write

比如你在代码里声明了一个todo工具：

{
    "name": "todo",
    "description": "Update task list. Track progress on multi-step tasks.",
    "input_schema": {
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                ...
            }
        },
        "required": ["items"]
    }
}

这相当于告诉模型：

你可以调用一个叫 todo 的工具，调用时需要传一个 items 参数。

于是模型如果想更新 todo，就可能返回一个 tool_use block，概念上类似：

{
    "type": "tool_use",
    "name": "todo",
    "input": {
        "items": [
            {"id": "1", "text": "阅读代码", "status": "completed"},
            {"id": "2", "text": "解释 TodoManager", "status": "in_progress"},
            {"id": "3", "text": "总结结果", "status": "pending"}
        ]
    }
}

这里的 input 就是模型生成的工具参数。

疑问1：

大模型怎么是按照任务清单逐步执行的？它怎么知道的哪个任务执行完了？这个任务是本地工具的任务还是大模型的任务？

简单总结：

大模型是靠 system prompt + todo 工具返回的状态 + 工具执行结果 + 自己的推理 来“看起来按任务清单逐步执行”的。

但是：

todo 不是自动任务队列，TodoManager 也不是调度器。
todo 里的任务是大模型的计划；
本地工具只执行具体动作；
哪个任务完成了，是大模型根据工具结果自己判断后，再主动更新 todo 状态。

大模型先根据用户目标，把事情拆成多个 todo，然后调用自定义的 todo 工具提交任务清单。这个任务清单会被本地的 TodoManager 保存、校验并渲染出来。

TodoManager 只负责三件事：

1. 保存当前 todo 列表；
2. 校验 todo 是否合法，比如状态只能是 pending、in_progress、completed，并且同一时间最多只能有一个 in_progress；
3. 把 todo 列表渲染成可读文本返回给模型。

真正执行任务的不是 TodoManager，而是大模型自己决定下一步该调用什么工具，比如 read_file、bash、edit_file 等。本地工具只负责执行具体动作，并返回结果。

任务是否完成，也不是本地工具自动判断的，而是大模型根据上下文和工具返回结果自己判断。比如读完文件后，模型认为“读取文件”这个 todo 完成了，就会再次调用 todo 工具，把该任务状态改成 completed，再把下一个任务改成 in_progress。

所以这里的 todo 更像是“进度记录”，帮助模型保持多步骤任务的方向，也方便用户观察执行进度。

rounds_since_todo 则是一个提醒计数器。如果模型连续几轮没有调用 todo 工具，代码会自动插入：

<reminder>Update your todos.</reminder>

提醒模型更新任务清单。

s04_subagent

疑问1

主Agent里调用subAgent应该是异步的吧？这里能立刻拿到output结果吗？

从“真实产品形态 / 多 Agent 调度”的角度看，主 Agent 调用 SubAgent 通常可以设计成异步；但在你这个 agents/s04_subagent.py 示例里：

output = run_subagent(prompt)

这里是同步阻塞调用，所以可以立刻拿到 output，但这个“立刻”不是瞬时返回，而是等 SubAgent 完整跑完之后才继续往下执行。

也就是说：

print(f"> task ({desc}): {prompt[:80]}")
output = run_subagent(prompt)
print(f"  {str(output)[:200]}")

执行顺序是：

1. 主 Agent 发现要调用 task 工具；
2. 打印 > task (...)；
3. 调用 run_subagent(prompt)；
4. 主 Agent 阻塞等待；
5. SubAgent 在 run_subagent 里独立循环调用模型、执行工具；
6. SubAgent 最终停止后，返回最后的文本摘要；
7. 主 Agent 才拿到 output；
8. 再把这个 output 作为 tool_result 加回主 Agent 上下文。

所以这份代码里的 SubAgent 调用是：

逻辑上是“委托子任务”，实现上是“同步等待子任务完成”。

疑问2

子Agent在哪里进行的信息summary？
简单总结：

run_subagent(prompt) 里，子 Agent 会独立完成任务。

在subAgent调用中，总结内容是由模型根据 system prompt 生成的，

SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."

意思是：

完成任务，然后总结发现。

所以流程是：

1. 主 Agent 调用 run_subagent(prompt)；
2. 子 Agent 用独立上下文执行任务；
3. 如果需要工具，就多轮调用工具；
4. 等子 Agent 不再调用工具时，模型会输出最终文本总结；
5. return "".join(...) 只是把这个最终文本提取出来；
6. 如果没有文本，就返回 "(no summary)"。

一句话概括：

总结摘要 是子 Agent 模型自己生成的，return "".join(...) 只是提取子 Agent 最后一次回复里的文本内容并返回给主 Agent。

s05_skill_loading

程序启动（python agents/s05_skill_loading.py）
  ↓
创建全局 SKILL_LOADER（SKILL_LOADER = SkillLoader(SKILLS_DIR)）
  ↓
扫描 skills 目录
  ↓
构造 SYSTEM prompt，只放 skill 简介
  ↓
用户提问
  ↓
模型发现需要某个 skill
  ↓
模型调用 load_skill 工具
  ↓
agent_loop 调用 SKILL_LOADER.get_content(name)
  ↓
完整 skill 内容作为 tool_result 返回给模型

s19_mcp

Agent 不再只能使用自己手写的工具，而是可以通过 MCP 这种标准协议，动态连接外部服务，发现外部工具，并把它们加入自己的工具池里使用。

也就是说，之前工具是：

Agent 内置：
bash
read_file
write_file
task
...

现在变成：

Agent 内置工具
+
外部 MCP Server 提供的工具

例如连接 docs 这个 MCP server 后，Agent 会多出：

mcp__docs__search
mcp__docs__get_version

如果没有 MCP，每接一个系统，你都要在 Agent 里手写一套工具逻辑。

例如：

Agent 里写 Jira 工具
Agent 里写部署工具
Agent 里写 Notion 工具
Agent 里写 GitLab 工具
...

这会让 Agent 越来越臃肿。

s19_mcp_plugin 的核心思想是：

Agent 只实现一套通用的 MCP Client。
外部服务自己实现 MCP Server。
Agent 通过标准协议发现和调用外部工具。

这一章的调用流程

以用户输入：

Connect to the docs MCP server and search for something

为例，流程是：

1. 用户输入 prompt

2. Agent 进入 agent_loop

3. assemble_tool_pool()
   此时只有内置工具

4. 模型决定调用 connect_mcp
   connect_mcp({"name": "docs"})

5. connect_mcp 找到 MOCK_SERVERS["docs"]

6. 创建 docs MCPClient

7. docs MCPClient 注册两个工具：
   - search
   - get_version

8. mcp_clients["docs"] = docs_client

9. agent_loop 检测到本轮调用了 connect_mcp

10. 重新 assemble_tool_pool()

11. 工具池里新增：
    - mcp__docs__search
    - mcp__docs__get_version

12. 模型继续调用：
    mcp__docs__search({"query": "something"})

13. handler 转发到：
    docs_client.call_tool("search", {"query": "something"})

14. mock handler 返回结果

这就是完整链路。

针对第5步中，当前是mock的服务数据，如果是真实的服务端，

connect_mcp("docs")
↓
从 MCP server 配置里找到 docs MCP Server 的连接信息
↓
建立连接：
  1. stdio：启动本地 MCP Server 子进程，用 stdin/stdout 通信 或
  2. HTTP / SSE / WebSocket：连接远程 MCP Server URL
↓
向 MCP Server 发送 tools/list
↓
获取 docs server 暴露的工具定义，比如：
  - search
  - get_version
↓
Agent 把这些工具改名并加入工具池：
  - mcp__docs__search
  - mcp__docs__get_version
↓
模型之后可以看到并选择调用这些工具
↓
当模型调用 mcp__docs__search 时
↓
Agent MCP Client 把调用转成 MCP 的 tools/call 请求
↓
Docs MCP Server 收到 tools/call
↓
Docs MCP Server 执行 search 这个工具的真实逻辑
↓
Docs MCP Server 把结果返回给 Agent
↓
Agent 把工具结果交回给模型
↓
模型基于结果继续回答用户

MCP、JSON-RPC、stdio / HTTP 三者关系

MCP 是协议；JSON-RPC 是 MCP 消息的编码格式；stdio / HTTP / SSE 是传输这些 JSON-RPC 消息的通道。

JSON-RPC 是什么？

JSON-RPC 是一种“用 JSON 表示远程过程调用”的格式，你可以把它理解成：

用统一 JSON 格式表达：我要调用哪个方法、参数是什么、返回结果是什么、有没有错误。

它不是 MCP 独有的东西，而是一种通用 RPC 消息格式。

JSON-RPC 长什么样？

请求 Request

比如 Agent 想问 MCP Server：

你有哪些工具？

MCP 语义是：

tools/list

JSON-RPC 消息可能长这样：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

含义：

jsonrpc: 使用 JSON-RPC 2.0
id: 这次请求的编号
method: 要调用的方法
params: 参数

---

响应 Response

Server 返回：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "search",
        "description": "Search documents",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string"
            }
          },
          "required": ["query"]
        }
      }
    ]
  }
}

这里的 id: 1 对应前面的请求 id: 1。

也就是说：

请求 id=1
响应 id=1

这样 Client 就知道这个响应是对应哪个请求的。

---

调用工具

比如 Agent 要调用文档搜索工具：

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": {
      "query": "MCP transport"
    }
  }
}

Server 返回：

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Found 3 documents about MCP transport."
      }
    ]
  }
}

疑问：为什么我写的MCP服务没有写tools/list方法？

#!/usr/bin/env node
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js';

// 代码里没有单独写tools/list的处理

tools/list 是 MCP 协议里的方法名；ListToolsRequestSchema 是 SDK 对这个方法的封装。你用 SDK 写 MCP Server 时，不需要直接写 "tools/list" 字符串，在SDK里已经帮你封装处理了。

在 MCP 里，Client 和 Server 会约定一些标准方法名，例如：

initialize
tools/list
tools/call
resources/list
prompts/list

这些方法名不是随便写的，而是协议规定的。

比如：

MCP 方法	含义
initialize	初始化连接，互相确认能力
tools/list	查看 Server 提供了哪些工具
tools/call	调用 Server 的某个工具
resources/list	查看 Server 提供了哪些资源
prompts/list	查看 Server 提供了哪些 prompt 模板

所以当我说：

MCP 语义是 tools/list

其实是在说：

这次请求在 MCP 协议层面的目的，是“列出工具”。

Egg.js中的MCPController

使用 MCPController 的好处是：

好处	说明
少写协议代码	不用自己实现 JSON-RPC、tools/list、tools/call
复用 Egg 能力	可以直接用 service、config、logger、数据库、插件
工具定义清晰	用装饰器声明 Tool、Prompt、Resource
Agent 更容易调用	description 和 schema 更规范
安全更好做	可以接入 Egg 的鉴权、权限、审计、限流
维护更方便	按 Controller 组织 MCP 工具
更适合已有项目	老 Egg 项目可以低成本接入 Agent

一句话：

MCPController 的价值是：把 “写一个 MCP Server” 这件事变成 “写一个 Egg Controller”，让你的 Egg 后端能力可以低成本、规范地暴露给 Agent 调用。