MCP 协议

pip install mcp

import asyncio

from mcp import Client
from mcp.server.mcpserver import MCPServer

server = MCPServer("weather")

@server.tool()
def get_weather(city: str) -> str:
    """查询城市当前的天气"""
    return f"{city}：晴"

async def main():
    async with Client(server) as client:
        result = await client.list_tools()
        tool = result.tools[0]
        print(tool.name)  # => get_weather
        print(tool.description)  # => 查询城市当前的天气
        print(tool.input_schema["required"])  # => ['city']

        answer = await client.call_tool("get_weather", {"city": "北京"})
        print(answer.content[0].text)  # => 北京：晴
        print(answer.is_error)  # => False

asyncio.run(main())

import asyncio

from mcp import Client
from mcp.server.mcpserver import MCPServer

server = MCPServer("docs")

@server.resource("docs://readme")
def readme() -> str:
    """项目说明"""
    return "这是项目说明"

@server.prompt()
def review(code: str) -> str:
    """审查一段代码"""
    return f"请审查这段代码：\n{code}"

async def main():
    async with Client(server) as client:
        resource = await client.read_resource("docs://readme")
        print(resource.contents[0].text)  # => 这是项目说明

        prompt = await client.get_prompt("review", {"code": "x = 1"})
        print(prompt.messages[0].role)  # => user

asyncio.run(main())

import asyncio
import sys

from mcp import Client, StdioServerParameters

async def main():
    params = StdioServerParameters(
        command=sys.executable,
        args=["weather_server.py"],
    )
    async with Client(params) as client:
        result = await client.call_tool("get_weather", {"city": "上海"})
        print(result.content[0].text)  # => 上海：晴

asyncio.run(main())

import asyncio
import json

from mcp import Client
from openai import AsyncOpenAI

llm = AsyncOpenAI()

def to_openai(tool) -> dict:
    return {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.input_schema,
        },
    }

async def run_agent(mcp: Client, task: str) -> str:
    tools = [to_openai(t) for t in (await mcp.list_tools()).tools]
    messages = [{"role": "user", "content": task}]
    for _ in range(10):
        response = await llm.chat.completions.create(
            model=MODEL, messages=messages, tools=tools,
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))
        if not message.tool_calls:
            return message.content
        for call in message.tool_calls:
            result = await mcp.call_tool(
                call.function.name,
                json.loads(call.function.arguments),
            )
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result.content[0].text,
            })
    raise RuntimeError("超过最大步数")

async def main():
    async with Client(server) as mcp:
        print(await run_agent(mcp, "北京天气？"))  # => 北京今天晴

asyncio.run(main())

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    }
  }
}