用 OpenAI Agents SDK 搭生产级 AI Agent（沙箱 + 工具 + MCP + 多 Agent 协作）

# 安装（Python 3.10+）
pip install "openai-agents>=0.14.0"

# 设置环境变量（bash / zsh）
export OPENAI_API_KEY="sk-..."

# 验证安装
python -c "import agents; print(agents.__version__)"

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(
        name="助理",
        instructions="你是技术助手，优先给可执行步骤。",
    )
    result = await Runner.run(agent, "给我一个 Python 项目接入 MCP 的最小步骤")
    print(result.final_output)

asyncio.run(main())

函数名会变成工具名，docstring 变成工具描述，类型注解变成参数 schema——你几乎零额外声明：

from typing import Annotated
from agents import Agent, Runner, function_tool

@function_tool
def query_kb(keyword: str) -> str:
    """查询知识库并返回摘要。"""
    fake_db = {"mcp": "MCP 用于让模型调用外部工具和资源。"}
    return fake_db.get(keyword.lower(), "未找到相关内容")

agent = Agent(
    name="研究助手",
    instructions="先调用 query_kb，再给 3 条实践建议。",
    tools=[query_kb],
)
result = Runner.run_sync(agent, "请解释 MCP")
print(result.final_output)

这是 v0.14 最关键的能力：Agent 在受控沙箱里读写文件、执行命令，不污染你的真机。内置 shell 和 apply_patch 工具，免手写：

from agents import Agent, Runner
from agents.sandbox import SandboxAgent

agent = SandboxAgent(
    name="代码助手",
    instructions="根据需求修改文件，然后运行测试验证。",
    sandbox={"backend": "local"},  # local / containerized / hosted
)
result = Runner.run_sync(
    agent,
    "在 /workspace 里创建 hello.py，写一个 add 函数和测试",
)
print(result.final_output)

from agents import Agent, Runner
from agents.mcp import MCPServerStdio

mcp_server = MCPServerStdio("python", args=["my_mcp_server.py"])
agent = Agent(
    name="MCP Agent",
    instructions="你可以调用 MCP 工具完成任务。",
    mcp_servers=[mcp_server],
)
result = Runner.run_sync(agent, "查询数据库中的用户列表")

from agents import Agent, Runner

research = Agent(name="研究", instructions="只做资料检索与综述。")
writer = Agent(name="写作", instructions="只把研究结果写成报告。")

router = Agent(
    name="协调",
    instructions="先交给研究 Agent，再 handoff 给写作 Agent。",
    handoffs=[research, writer],
)
result = Runner.run_sync(router, "调研 Agent 框架趋势并出一份简报")

from agents import Agent, Runner, input_guardrail

@input_guardrail
async def no_prompt_injection(ctx, agent, inp):
    # 简单示例：拦截含「忽略以上指令」的输入
    if "忽略" in inp:
        raise GuardrailTripwire("检测到注入尝试")

agent = Agent(name="安全助手", instructions="...", input_guardrails=[no_prompt_injection])
# 默认开启 tracing，按 run_id 追踪每一步