目录

1. 简介
2. 环境准备与依赖安装
3. 创建最简单的服务器端工具
4. 客户端连接与调用
5. 异步编程模式详解
6. 常见问题排查
7. 总结

简介

本指南旨在帮助具备Python基础的开发者在10分钟内快速搭建并运行第一个MCP（Model Context Protocol）应用。我们将通过simple-tool示例创建一个服务器端工具，通过simple-chatbot示例创建一个客户端会话来调用该工具。整个过程将涵盖从依赖安装、服务器创建、客户端连接到工具调用的完整端到端流程。

环境准备与依赖安装

要开始使用MCP SDK，首先需要通过pyproject.toml文件安装SDK及其所有依赖。此文件定义了项目所需的所有核心依赖项。

[project]
name = "mcp"
requires-python = ">=3.10"
dependencies = [
    "anyio>=4.5",
    "httpx>=0.27.1",
    "httpx-sse>=0.4",
    "pydantic>=2.11.0,<3.0.0",
    "starlette>=0.27",
    "python-multipart>=0.0.9",
    "sse-starlette>=1.6.1",
    "pydantic-settings>=2.5.2",
    "uvicorn>=0.31.1; sys_platform != 'emscripten'",
    "jsonschema>=4.20.0",
    "pywin32>=310; sys_platform == 'win32'",
]

安装步骤：

1. 确保已安装Python 3.10或更高版本。
2. 使用pip安装MCP SDK：

   pip install mcp
   

   此命令将自动解析并安装pyproject.toml中定义的所有依赖项。

本节来源

• pyproject.toml

创建最简单的服务器端工具

我们将基于simple-tool示例创建一个最简单的服务器端工具。该服务器将注册一个名为fetch的基本函数，用于获取指定URL的网页内容。

服务器核心结构

服务器的核心是一个Server实例，它负责管理工具的注册和调用。我们使用@app.call_tool()和@app.list_tools()装饰器来注册工具。

from mcp.server.lowlevel import Server

app = Server("mcp-website-fetcher")

注册基本函数

我们首先定义一个异步函数fetch_website，它使用httpx库获取网页内容并返回一个TextContent对象。

async def fetch_website(url: str) -> list[types.ContentBlock]:
    headers = {"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"}
    async with create_mcp_http_client(headers=headers) as client:
        response = await client.get(url)
        response.raise_for_status()
        return [types.TextContent(type="text", text=response.text)]

然后，我们使用@app.call_tool()装饰器注册一个工具调用处理器fetch_tool，它会验证输入参数并调用fetch_website函数。

@app.call_tool()
async def fetch_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
    if name != "fetch":
        raise ValueError(f"Unknown tool: {name}")
    if "url" not in arguments:
        raise ValueError("Missing required argument 'url'")
    return await fetch_website(arguments["url"])

暴露工具（通过FastMCP）

为了让客户端发现可用的工具，我们需要实现list_tools方法。该方法返回一个Tool对象列表，每个对象都包含工具的名称、描述和输入参数的JSON Schema。

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="fetch",
            title="Website Fetcher",
            description="Fetches a website and returns its content",
            inputSchema={
                "type": "object",
                "required": ["url"],
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL to fetch",
                    }
                },
            },
        )
    ]

启动服务器

最后，我们通过命令行参数选择传输方式（stdio或sse）并启动服务器。对于SSE（Server-Sent Events）传输，我们使用uvicorn和Starlette框架来创建一个HTTP服务器。

if transport == "sse":
    # ... SSE服务器设置
    uvicorn.run(starlette_app, host="127.0.0.1", port=port)
else:
    # ... stdio服务器设置
    anyio.run(arun)

本节来源

• examples/servers/simple-tool/mcp_simple_tool/server.py

客户端连接与调用

现在，我们将创建一个客户端来连接到上述服务器并调用fetch工具。我们参考simple-chatbot示例来实现这一过程。

配置与初始化

客户端首先通过Configuration类加载环境变量和服务器配置。servers_config.json文件定义了要连接的服务器的命令、参数和环境变量。

{
  "mcpServers": {
    "website-fetcher": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/sdk-js-examples", "simple-tool"],
      "env": {}
    }
  }
}

建立会话

客户端使用ClientSession类来建立与服务器的会话。Server类的initialize方法负责启动服务器进程并建立stdio通信通道。

async def initialize(self) -> None:
    server_params = StdioServerParameters(
        command=command,
        args=self.config["args"],
        env={**os.environ, **self.config["env"]} if self.config.get("env") else None,
    )
    try:
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        read, write = stdio_transport
        session = await self.exit_stack.enter_async_context(ClientSession(read, write))
        await session.initialize()
        self.session = session
    except Exception as e:
        logging.error(f"Error initializing server {self.name}: {e}")
        await self.cleanup()
        raise

调用工具

一旦会话建立，客户端就可以通过session.call_tool()方法调用服务器上注册的工具。

async def execute_tool(
    self,
    tool_name: str,
    arguments: dict[str, Any],
    retries: int = 2,
    delay: float = 1.0,
) -> Any:
    if not self.session:
        raise RuntimeError(f"Server {self.name} not initialized")

    attempt = 0
    while attempt < retries:
        try:
            logging.info(f"Executing {tool_name}...")
            result = await self.session.call_tool(tool_name, arguments)
            return result
        except Exception as e:
            # ... 重试逻辑
            raise

本节来源

• examples/clients/simple-chatbot/mcp_simple_chatbot/main.py
• examples/clients/simple-chatbot/mcp_simple_chatbot/servers_config.json

异步编程模式详解

MCP SDK广泛使用async/await异步编程模式，这对于处理I/O密集型操作（如网络请求）至关重要。

关键异步API调用

• async with: 用于管理异步上下文管理器，确保资源被正确清理。

  async with create_mcp_http_client(headers=headers) as client:
      response = await client.get(url)
  

• await: 用于等待异步操作完成，如网络请求或会话初始化。

  await session.initialize()
  result = await self.session.call_tool(tool_name, arguments)
  

• asyncio.run(): 作为程序的入口点，启动异步事件循环。

  def main() -> None:
      asyncio.run(run())
  

参数含义

• validate_input: 在@app.call_tool()装饰器中，此参数控制是否根据inputSchema验证输入参数，默认为True。
• raise_exceptions: 在Server.run()方法中，此参数控制异常处理方式。True时异常会中断服务器，False时异常会作为消息返回给客户端。

常见问题排查

端口占用

如果使用SSE模式，确保指定的端口（默认8000）未被其他进程占用。可以通过--port参数更改端口：

python server.py --transport sse --port 8080

依赖缺失

如果遇到ModuleNotFoundError，请确保已正确安装所有依赖：

pip install -e .

或者，如果使用uv工具：

uv sync

环境变量

确保LLM_API_KEY等必要的环境变量已设置。客户端示例使用.env文件来管理这些变量。

本节来源

• examples/servers/simple-tool/mcp_simple_tool/server.py
• examples/clients/simple-chatbot/mcp_simple_chatbot/main.py

总结

通过本指南，您已成功完成了MCP应用的端到端搭建。您学会了如何安装SDK依赖、创建一个简单的服务器端工具、通过FastMCP暴露该工具，以及使用客户端会话连接并调用该工具。关键的异步编程模式和API参数也已详细解释。现在，您可以基于此基础构建更复杂的应用。

本节来源

• pyproject.toml
• examples/servers/simple-tool/mcp_simple_tool/server.py
• examples/clients/simple-chatbot/mcp_simple_chatbot/main.py