1

pip install "mcp[cli]"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# weather_server.py
import httpx
from mcp.server.fastmcp import FastMCP

# 创建 MCP Server
mcp = FastMCP("Weather Service", json_response=True)

# 定义工具：获取天气预报
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取指定坐标的天气预报"""
    url = f"https://api.weather.gov/points/{latitude},{longitude}"
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers={"User-Agent": "mcp-demo/1.0"})
        resp.raise_for_status()
        forecast_url = resp.json()["properties"]["forecast"]
        forecast_resp = await client.get(forecast_url)
        forecast_resp.raise_for_status()
        return forecast_resp.json()["properties"]["periods"][:3]

# 定义工具：获取天气预警
@mcp.tool()
async def get_alerts(state: str) -> list[dict]:
    """获取指定州的天气预警"""
    url = f"https://api.weather.gov/alerts/active/area/{state}"
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers={"User-Agent": "mcp-demo/1.0"})
        resp.raise_for_status()
        data = resp.json()
        return [
            {"event": a["properties"]["event"], "headline": a["properties"]["headline"]}
            for a in data.get("features", [])[:5]
        ]

# 定义资源：暴露静态数据
@mcp.resource("weather://supported-states")
async def get_supported_states() -> str:
    """返回支持的美国州代码列表"""
    return "AL, AK, CA, NY, TX, FL, WA, OR"

if __name__ == "__main__":
    mcp.run(transport="stdio")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 开发模式（带 MCP Inspector 调试 UI）
uv run mcp dev weather_server.py

# 直接运行（stdio 传输）
python weather_server.py

# Streamable HTTP 传输
python -c "
from weather_server import mcp
mcp.run(transport='streamable-http')
"

1

pip install a2a-sdk

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# a2a_translation_server.py
from a2a_sdk.server import AgentCard, A2AServer
from a2a_sdk.types import (
    Task, TaskState, TaskStatus, Message, TextContent,
    AgentCard as AgentCardModel,
)

class TranslationAgent:
    """翻译 Agent，支持中英文互译"""

    async def get_agent_card(self) -> AgentCardModel:
        return AgentCardModel(
            name="Translation Agent",
            description="中英文翻译服务",
            url="http://localhost:8080",
            version="1.0.0",
            capabilities={
                "translation": {
                    "source_languages": ["zh", "en"],
                    "target_languages": ["zh", "en", "ja", "ko", "fr"],
                }
            },
        )

    async def handle_task(self, task: Task) -> Task:
        """处理翻译任务"""
        # 从任务消息中提取待翻译文本
        message = task.messages[-1]
        text = message.content.text

        # 简单翻译逻辑（生产环境应调用 LLM 或翻译 API）
        translations = {
            "hello": "你好",
            "world": "世界",
            "你好": "Hello",
            "世界": "World",
        }
        translated = translations.get(text.strip().lower(), f"[翻译]{text}")

        task.status = TaskStatus(state=TaskState.COMPLETED)
        task.messages.append(
            Message(
                role="agent",
                content=TextContent(text=translated),
            )
        )
        return task

# 启动 A2A Server
if __name__ == "__main__":
    import uvicorn
    from a2a_sdk.server import create_app

    agent = TranslationAgent()
    app = create_app(agent, host="0.0.0.0", port=8080)
    uvicorn.run(app, host="0.0.0.0", port=8080)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# a2a_client_example.py
from a2a_sdk.client import A2AClient
from a2a_sdk.types import Task, TaskState, Message, TextContent

async def translate_via_a2a():
    """通过 A2A 协议调用翻译 Agent"""
    client = A2AClient(base_url="http://localhost:8080")

    # 1. 获取 Agent Card（发现能力）
    card = await client.get_agent_card()
    print(f"发现 Agent: {card.name}")
    print(f"能力: {list(card.capabilities.keys())}")

    # 2. 发送翻译任务
    task = Task(
        messages=[
            Message(
                role="user",
                content=TextContent(text="hello"),
            )
        ]
    )
    result = await client.send_task(task)

    # 3. 轮询任务结果
    while result.status.state != TaskState.COMPLETED:
        result = await client.get_task(result.id)

    print(f"翻译结果: {result.messages[-1].content.text}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(translate_via_a2a())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# multi_agent_demo.py
import asyncio
from agents import Agent, Runner, set_trace_processors
from agents.tracing.processors import ConsoleTracingProcessor

# 启用控制台追踪
set_trace_processors([ConsoleTracingProcessor()])

# 定义三个专业 Agent
triage_agent = Agent(
    name="Triage Agent",
    instructions="你是客服分流 Agent。根据用户的问题类型，将任务转交给对应的专业 Agent。",
    handoffs=["billing_agent", "tech_support_agent"],
)

billing_agent = Agent(
    name="Billing Agent",
    instructions="你是账单 Agent。回答用户关于账单、发票和支付的问题。",
)

tech_support_agent = Agent(
    name="Tech Support Agent",
    instructions="你是技术支持 Agent。帮助用户解决产品使用问题和技术故障。",
)

# 也可以用函数作为工具
async def get_account_info(account_id: str) -> dict:
    """获取用户账户信息"""
    return {
        "account_id": account_id,
        "name": "张三",
        "plan": "pro",
        "balance": 199.00,
    }

billing_agent.tools = [get_account_info]

async def main():
    # 运行 Agent
    result = await Runner.run(
        triage_agent,
        input="我的账单上个月扣了 199 元，能帮我查一下吗？",
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# sandbox_agent_demo.py
from agents import Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.entries import GitRepo
from agents.sandbox.sandboxes import UnixLocalSandboxClient

agent = SandboxAgent(
    name="Code Reviewer",
    instructions="检查克隆下来的仓库代码，发现 Bug 并修复。",
    default_manifest=Manifest(
        entries={
            "my_project": GitRepo(
                repo="openai/openai-agents-python",
                ref="main",
            ),
        }
    ),
)

result = Runner.run_sync(
    agent,
    "检查这个项目的 README，并告诉我它解决什么问题。",
    run_config=RunConfig(
        sandbox=SandboxRunConfig(client=UnixLocalSandboxClient())
    ),
)
print(result.final_output)