搬家前：工具住在 agent 进程里
对应清单 3.8 · 这一版没有 MCP
只有一个进程
05_time_travel_agent.py
agent + 工具 + _journal 全在这儿
进程内调用
两个工具：记一笔、把日记读出来。日记就是个内存里的 _journal 列表。
纯第二章的写法，@function_tool 而已。
import asyncio
from agents import (Agent, Runner, function_tool,
set_default_openai_api, set_tracing_disabled)
from dotenv import load_dotenv
# 仓库原文没有这三行，是为了接第三方模型加的
load_dotenv()
set_default_openai_api("chat_completions")
set_tracing_disabled(True)
# 内存里的日记状态（一个列表）
_journal = []
@function_tool
def record_event(entry: str) -> dict:
"""把一条新的旅行事件记进日记。"""
_journal.append(entry)
print(f"已记录事件：{entry}")
return {"status": "recorded", "entry": entry}
@function_tool
def load_journal() -> dict:
"""读出当前日记里的所有条目。"""
print("正在读取日记条目……")
return {"status": "loaded", "journal": "\n".join(_journal)}
agent = Agent(
name="时光记录员",
instructions="""你是一个时光旅行日记智能体。
每次开始时都要先用 'load_journal' 工具把过去的条目取出来。
遇到新事件，调用 'record_event' 把它存下来。
如果用户要看总结或日记，就把所有记录过的事件输出来。""",
tools=[record_event, load_journal],
)
# 模拟一串历史旅行事件
travel_events = [
"去古罗马看了一场角斗士比赛",
"见证了 1776 年《独立宣言》的签署",
"目睹了 1969 年的登月",
]
async def main():
print("正在记录旅行：")
for event in travel_events:
await Runner.run(agent, event)
# 让智能体总结一下这些冒险
result = await Runner.run(agent, "把我的旅行历史给我看看")
print("\n最终日记：")
print(result.final_output)
asyncio.run(main())
instructions 里的工具名不能翻
提示词里那两个 'load_journal' 和 'record_event' 保持英文，
因为它们必须和函数名一字不差——模型是照着这个名字去调工具的。
翻成中文，模型就会去找一个根本不存在的工具。
搬家：同样的工具，换个装饰器就出了进程
对应清单 3.9
把上面两个工具原样搬进一个 MCP server。函数体一个字没改，
只是 @function_tool → @mcp.tool()：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Time Travel Tracker")
# 内存里的日记状态（一个列表）
_journal = []
@mcp.tool()
def record_event(entry: str) -> dict:
"""把一条新的旅行事件记进日记。"""
_journal.append(entry)
print(f"已记录事件：{entry}")
return {"status": "recorded", "entry": entry}
@mcp.tool()
def load_journal() -> dict:
"""读出当前日记里的所有条目。"""
print("正在读取日记条目……")
return {"status": "loaded", "journal": "\n".join(_journal)}
if __name__ == "__main__":
mcp.run(transport="sse")
但是有件大事悄悄变了：_journal 现在住在别的进程里。
前一版里它是你 Python 脚本的一个局部变量；这一版它属于 server 进程。
这带来一个第二章不可能遇到的现象——状态是共享的。
两个 agent 同时连上同一个 SSE server，会往同一本日记里写。
书里也提到了这点：你可能会看到重复的日志，因为 server 进程不是每次都重建的。
工具里那两个 print，是个定时炸弹
print() 写的是标准输出，而 STDIO 传输的 JSON-RPC 走的也是标准输出——
同一个 fd。我抓过原始字节，确认了 print 的内容真的会混进协议流：
已记录事件：去古罗马 ← 不是 JSON
{"jsonrpc":"2.0","id":2,"result":…}
那为什么跑起来没事？因为现在的 Python MCP 客户端够宽容，遇到解析不了的行就跳过。
这是在赌运气，不是设计如此。我实测过：让工具狂 print 几百行，
管道被写满，整个调用直接死锁挂起，再也不返回。换个别的客户端实现
（Claude 桌面端、TypeScript SDK）也可能直接断连。
正确做法：MCP server 里要打日志，就往 stderr 打
（print(..., file=sys.stderr)），别占 stdout。
这一版是 SSE 跑的，stdout 空着，所以确实没事——
可下一节那个 stdio 客户端，跑的正是这同一个文件。
搬家后：agent 这边几乎没变
对应清单 3.10 / 3.11
进程 A
06_time_travel_agent_mcp_stdio.py
只剩提示词，没有工具代码了
STDIO
进程 B
06_mcp_time_travel_tracker.py
工具 + _journal 都在这边
对比一下上一版：工具函数没了，_journal 没了，
提示词一字未动，tools= 变成了 mcp_servers=。
agent 完全不知道日记存在哪、谁在管——它只知道对面有两个工具可以调。
import asyncio
from pathlib import Path
from agents import Agent, Runner, set_default_openai_api, set_tracing_disabled
from agents.mcp import MCPServerStdio, MCPServerStdioParams
from dotenv import load_dotenv
# 仓库原文没有这三行，是为了接第三方模型加的
load_dotenv()
set_default_openai_api("chat_completions")
set_tracing_disabled(True)
SCRIPT = Path(__file__).with_name("06_mcp_time_travel_tracker.py").resolve()
# 模拟一串历史旅行事件
travel_events = [
"去古罗马看了一场角斗士比赛",
"见证了 1776 年《独立宣言》的签署",
"目睹了 1969 年的登月",
]
async def main():
async with MCPServerStdio(
name="Time Tracker Server",
params=MCPServerStdioParams(
command="mcp",
args=["run", str(SCRIPT)],
),
) as time_tracker_server:
agent = Agent(
name="助手",
instructions="""
你是一个时光旅行日记智能体。
每次开始时都要先用 'load_journal' 工具把过去的条目取出来。
遇到新事件，调用 'record_event' 把它存下来。
如果用户要看总结或日记，就把所有记录过的事件输出来。
""",
mcp_servers=[time_tracker_server],
)
print("正在记录旅行：")
for event in travel_events:
await Runner.run(agent, event)
# 让智能体总结一下这些冒险
result = await Runner.run(agent, "把我的旅行历史给我看看")
print("\n最终日记：")
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
SSE 版就是把 MCPServerStdio 换成 MCPServerSse、
改成填 URL。但这一版你得先手动开 server——而书在这儿给错了命令，
详见下面「踩坑」第一条。
import asyncio
from agents import Agent, Runner, set_default_openai_api, set_tracing_disabled
from agents.mcp import MCPServerSse
from dotenv import load_dotenv
# 仓库原文没有这三行，是为了接第三方模型加的
load_dotenv()
set_default_openai_api("chat_completions")
set_tracing_disabled(True)
# 模拟一串历史旅行事件
travel_events = [
"去古罗马看了一场角斗士比赛",
"见证了 1776 年《独立宣言》的签署",
"目睹了 1969 年的登月",
]
async def main():
async with MCPServerSse(
name="Time Tracker Server",
params={
"url": "http://localhost:8000/sse",
},
) as time_tracker_server:
agent = Agent(
name="助手",
instructions="""
你是一个时光旅行日记智能体。
每次开始时都要先用 'load_journal' 工具把过去的条目取出来。
遇到新事件，调用 'record_event' 把它存下来。
如果用户要看总结或日记，就把所有记录过的事件输出来。
""",
mcp_servers=[time_tracker_server],
)
print("正在记录旅行：")
for event in travel_events:
await Runner.run(agent, event)
# 让智能体总结一下这些冒险
result = await Runner.run(agent, "把我的旅行历史给我看看")
print("\n最终日记：")
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
这个文件我删掉了仓库里的一行
仓库原文第 8 行有这么一句：
SCRIPT = Path(__file__).with_name("06_mcp_time_travel_tracker").resolve()。
它是死代码——SSE 版靠 URL 连接，根本用不到 SCRIPT。而且它还写错了，
少了 .py 后缀。大概是从 stdio 版复制过来忘了删。上面这版我去掉了。
这一节真正想让你看见的
把 05 和 06_..._stdio 并排放，agent 那部分的差别只有两行：
tools=[...] → mcp_servers=[...]，以及外面套了个 async with。
提示词、事件列表、调用方式，全都没动。
代价是多了一个进程、多了一层协议、调试难度上了一个台阶。
换来的是：这套工具现在能被任何 agent 复用、能被 Claude 桌面端直接用、
能换成任何语言重写而调用方毫无感知。这笔账划不划算，取决于这些工具会不会被第二个人用。
只有你自己用，@function_tool 挺好。