MCP 基础概念

MCP github 主页， MCP 官方文档

MCP Server

# server.py
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

# Create an MCP server
mcp = FastMCP("Demo")


# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

安装对应 python 环境： pip install "mcp[cli]" ，运行 mcp dev server.py 。

打开对应的 MCP Inspector 进行测试：command 填 python，arguments 填 server.py。点击链接后出现如下界面。

image-20250620211847064

在该界面上可以测试 list_tools， 调用 tool 等方式，MCP 采用了 JSON-RPC 2.0 作为消息格式，实例数据格式：

{method: "tools/call", params: {name: "add", arguments: {a: 1, b: 0}, _meta: {progressToken: 6}},…}

MCP Client

建立连接

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
    # methods will go here
    
    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server

        Args:
            server_script_path: Path to the server script (.py or .js)
        """
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("Server script must be a .py or .js file")

        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )

        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        await self.session.initialize()

        # List available tools
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])
        
        # call tool example 
        # result = await self.session.call_tool(tool_name, tool_args)
        
async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        # await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    import sys
    asyncio.run(main())
    # python client.py server.py

MCP Client 的 python SDK 提供了连接到 mcp server 的函数，如上。

Tool

详细参考官方文档

在建立好连接后，可以使用 ClientSession 一类来获取对应的 Tool 信息，如

# client.py
async def tool_demo(self):
    # get tool info
    response = await self.session.list_tools()
    tools = response.tools
    print("\nConnected to server with tools:", [tool for tool in tools])

    # call tool example 
    result = await self.session.call_tool("add", {"a":1, "b":2})
    print(f"{result=}")

运行 python client.py server.py 可以从 mcp server 那边执行 tool_demo 函数，打印结果：

Connected to server with tools: [Tool(name='add', description='Add two numbers', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'}, annotations=None)]
result=CallToolResult(meta=None, content=[TextContent(type='text', text='3', annotations=None)], isError=False)

MCP server 中规定的 tool 包含了以下信息：

{
  name: string;          // Unique identifier for the tool
  description?: string;  // Human-readable description
  inputSchema: {         // JSON Schema for the tool's parameters
    type: "object",
    properties: { ... }  // Tool-specific parameters
  },
  annotations?: {        // Optional hints about tool behavior
    title?: string;      // Human-readable title for the tool
    readOnlyHint?: boolean;    // If true, the tool does not modify its environment
    destructiveHint?: boolean; // If true, the tool may perform destructive updates
    idempotentHint?: boolean;  // If true, repeated calls with same args have no additional effect
    openWorldHint?: boolean;   // If true, tool interacts with external entities
  }
}

prompt

也可以从 MCP 服务器提取一些 prompt：

async def prompt_demo(self):
    prompts = await self.session.list_prompts()
    print(f"{prompts=}")
    for prompt in prompts.prompts:
        p = await self.session.get_prompt(prompt.name, arguments={"code": f"this is a code for {prompt.name}"})
        print(f"\nPrompt {p=}")

返回的 prompt 结果 p 通常为一个 list，其中包含了 role，content， annotation 等信息。

MCP server 中的 prompt 都通过一下格式定义：

{
  name: string;              // Unique identifier for the prompt
  description?: string;      // Human-readable description
  arguments?: [              // Optional list of arguments
    {
      name: string;          // Argument identifier
      description?: string;  // Argument description
      required?: boolean;    // Whether argument is required
    }
  ]
}

resources

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.resource("config://app", title="Application Configuration")
def get_config() -> str:
    """Static configuration data"""
    return "App configuration here"


@mcp.resource("users://{user_id}/profile", title="User Profile")
def get_user_profile(user_id: str) -> str:
    """Dynamic user data"""
    return f"Profile data for user {user_id}"

resource 可以返回字符串，dict，或 bytes 类型的数据，但 resource 中需要知名对应的 mime_type， 如 "image/png", "application/octet-stream" 等

参考资源

官方 server github

fastmcp