简介

MCP（Model Control Protocol） 是一种基于 JSON-RPC 2.0 协议的通信协议，主要用于在 AI 模型前端客户端 和 后端服务 之间建立和维护连接。它的目标是提供一种统一、结构化、标准化的方式，来控制和管理模型交互流程

交互流程

1. 客户端打开一个 SSE 连接

curl -N http://localhost:8000/sse

服务端会返回endpoint事件：

event:endpoint
data:/mcp/message?sessionId=a0a77dd093634b009d9f90d1996a9f1b

注意：

• SSE (HTTP/1.1)只能单向通信，连接建立后，服务端流式推送消息到客户端。
• 客户端每次建立sse连接，endpoint都会带上标识本次会话的sessionId。
• MCP利用endpoint实现了伪双工，客户端推送消息到服务端利用POST 的方式。

2. 客户端发送initialize请求

curl -X POST "http://localhost:8000/messages/?session_id=a0a77dd093634b009d9f90d1996a9f1b" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 0,
    "method": "initialize",
    "params": {"capabilities": {"roots": {"listChanged": true}, "sampling": {}}, "clientInfo": {"name": "mcp", "version": "0.1.0"}, "protocolVersion": "2024-11-05"}
  }'

3. 服务端响应initialize请求

服务端通过之前建议的sse端点，向客户端答复已经接受到initialize请求

event: message
data: {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2024-11-05","capabilities":{"experimental":{},"tools":{"listChanged":false}},"serverInfo":{"name":"hello-app","version":"1.8.1"}}}

4. 客户端发送initialized通知

在mcp协议中，通知是one-way消息，接收方不会做任何响应。

curl -X POST "http://localhost:8000/messages/?session_id=a0a77dd093634b009d9f90d1996a9f1b" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "notifications/initialized"
  }'

在经过这三次握手后，客户端与服务端的连接处于已建立的状态，后续可以实际请求：

5. 获取工具列表

客户端在连接建立后，可以发送一条listTool消息，获取server支持的工具列表：

curl -X POST "http://localhost:8000/messages/?session_id=a0a77dd093634b009d9f90d1996a9f1b" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list"
  }'

服务端收到消息后，会通过sse端点返回结果：

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"say_hello","description":"Returns a friendly greeting","inputSchema":{"type":"object","required":["name"],"properties":{"name":{"type":"string","description":"The name to greet"}}}}]}}

6. 调用指定工具

客户端可以发送一条tool_call消息，调用指定的工具

curl -X POST "http://localhost:8000/messages/?session_id=a0a77dd093634b009d9f90d1996a9f1b" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
	"params": { "name": "say_hello", "arguments": {"name": "world"} }
  }'

服务端收到消息后，会通过sse端点返回结果：

event: message
data: {"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"hello world"}],"isError":false}}

SDK

MCP官方提供常见开发语言的SDK，帮助你快速暴露MCP服务，或者调用MCP工具，详细见Model Context Protocol

Python

该示例提供了一个基础版本的hello world，使用SSE协议。
依赖安装：pip install mcp

mcp_server.py

import logging

import click
import mcp.types as types
import uvicorn
from mcp.server.lowlevel import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Mount, Route

logger = logging.getLogger(__name__)


@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
def main(
    port: int,
) -> int:
    app = Server("hello-app")

    @app.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:

        async def say_hello(args: dict):
            return [
                types.TextContent(
                    type="text",
                    text=f"hello {args.get('name', '')}",
                )
            ]

        if name == 'say_hello':
            return await say_hello(arguments)

        raise ValueError(f"Tool not found: {name}")


    @app.list_tools()
    async def list_tools() -> list[types.Tool]:
        return [
            types.Tool(
                name="say_hello",
                description="Returns a friendly greeting",
                inputSchema={
                    "type": "object",
                    "required": ["name"],
                    "properties": {
                        "name": {
                            "type": "string",
                            "description": "The name to greet",
                        },
                    },
                },
            )
        ]

    sse = SseServerTransport("/messages/")

    async def handle_sse(request):
        async with sse.connect_sse(
                request.scope, request.receive, request._send
        ) as streams:
            await app.run(
                streams[0], streams[1], app.create_initialization_options()
            )
        return Response()

    starlette_app = Starlette(
        debug=True,
        routes=[
            Route("/sse", endpoint=handle_sse, methods=["GET"]),
            Mount("/messages/", app=sse.handle_post_message),
        ],
    )

    uvicorn.run(starlette_app, host="0.0.0.0", port=port)


if __name__ == "__main__":
    main()

mpc_client.py

from mcp.client.session import ClientSession
from mcp.client.sse import sse_client


async def main():
    try:
        async with sse_client("http://localhost:8000/sse") as streams:
            async with ClientSession(
                streams[0],
                streams[1],
            ) as session:
                await session.initialize()
                tools = await session.list_tools()
                print(f"{tools.tools=}")
                response = await session.call_tool(name="say_hello", arguments={"name": "world"})
                print(f"Tool response: {response.model_dump()}")

    except Exception as e:
        print(e)


if __name__ == "__main__":
    from anyio import run

    run(main)