目录

  1. 核心概念 plus
    1. Transport 类型
    2. Tools
    3. Resources
    4. Prompts
    5. Sampling
    6. Roots
  2. END

上一篇重点讲了什么是 mcp、为何使用 mcp 以及如何使用。本来想把概念一起都写到那里，但是觉得太长了，干脆就把概念介绍放到了这里。

核心概念 plus

Transport 类型

刚才我们在 demo 里用的是 transport="streamable-http" ，实际上 transport 有 3 种类型：

  • stdio（默认）
  • sse
  • streamable-http

第一种是命令行启动 server 程序（不是 http），比如 uv --directory /ABSOLUTE/PATH/TO/PARENT/FOLDER/get_current_time run get_current_time.py ，实际上就是直接运行 python 程序。你填入 inspector 或者 postman 的时候就是填入启动命令。

后两种都是 http 形式的，最终会暴露出来一个 url 来提供调用，sse 默认的端点是 /sse ，streamable-http 默认是 /mcp ，而且根据文档，后者已经在逐渐替换前者了，所以建议使用后者进行开发。两者区别如下：

SSE streamable-http
endpoint /sse /mcp
启动参数 transport="sse" transport="streamable-http"
推荐场景 兼容老客户端、本地开发 生产环境、大并发、云部署
是否支持流式 支持（单向） 支持（更灵活，多格式）
多节点/云原生 不友好 非常友好
负载均衡 难 易
Session管理 简单/无状态 支持 stateful/stateless/resume

Tools

如果用过 openai 的 function calling，那么应该对这个很熟了，本质上就是将函数定义以一定格式传给模型。

工具由如下结构定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
{
  name: string;          // 工具的唯一标识符
  description?: string;  // 易读的工具描述
  inputSchema: {         // 工具参数的 JSON schema
    type: "object",
    properties: { ... }  // 工具参数
  },
  annotations?: {        // 可选，关于工具行为的注释
    title?: string;      // 易读的工具标题
    readOnlyHint?: boolean;    // 如果 true，那么工具不会改变其环境（只读）。
    destructiveHint?: boolean; // 如果 true，那么工具可能会执行破坏性的更新。
    idempotentHint?: boolean;  // 如果 true，那么对该工具使用同样参数重复调用，不会有额外影响。
    openWorldHint?: boolean;   // 如果 true，那么工具可能会与外部实体交互。
  }
}

这么说可能有点抽象，我们来看下刚才的 get_current_time 工具的定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
{
  "tools": [
    {
      "name": "get_current_time",
      "description": "\n    Returns the current date and time in the 'Asia/Shanghai' timezone.\n\n    Returns:\n        datetime: The current datetime object localized to 'Asia/Shanghai' timezone.\n    ",
      "inputSchema": {
        "type": "object",
        "properties": {},
        "title": "get_current_timeArguments"
      }
    }
  ]
}

其他还有执行系统命令和调用 api 的工具等例子，可以参考这里。

我们在使用调试工具连接 mcp server 后，会自动发起 ListToolsRequest 请求来获取可用的工具：

Resources

资源就是 server 希望提供给 client 的数据，包括文本数据和二进制数据：

  • 文本数据：utf-8 编码的文本数据，例如代码、日志、JSON 和纯文本等。
  • 二进制数据：PDF、图像、音频和视频等。

每个资源都有一个唯一标识 URI，格式为 [protocol]://[host]/[path]，比如我们常见的 pg 数据库可以标识为 postgres://database/customers/schema ，文件可以标识为 file:///home/user/documents/report.pdf 。另外 server 也可以定义自己的 URI 格式。

server 在返回资源列表时，每个资源一般使用如下格式返回：

1
2
3
4
5
6
{
  uri: string;           // 资源 URI
  name: string;          // 人类可读的名字
  description?: string;  // 可选，资源描述，这个会提供给 llm，使其更好的理解该资源。
  mimeType?: string;     // 可选，MIME type，如v text/plain
}

如果一个资源的 URI 是动态的，那么上述 uri 字段可以换成 uriTemplate ，其格式遵循 RFC 6570。

client 可以通过 resources/read 来读取可用的资源，server 返回资源列表。另外在资源更新时 server 也可以通知 client，具体可以见 Resource updates。

一个实现例子：

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
app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///logs/app.log",
            name="Application Logs",
            mimeType="text/plain"
        )
    ]

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    if str(uri) == "file:///logs/app.log":
        log_contents = await read_log_file()
        return log_contents

    raise ValueError("Resource not found")

# Start server
async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )

Prompts

在 mcp 中，prompts 其实指的是 prompt template，即带有参数的 prompt，说白了就是 python 中的 f-string。

这个 template 的定义如下：

1
2
3
4
5
6
7
8
9
10
11
{
  name: string;              // prompt 的唯一标识符
  description?: string;      // 人类可读的描述
  arguments?: [              // 可选，参数列表
    {
      name: string;          // 参数标识符
      description?: string;  // 参数描述
      required?: boolean;    // 该参数是否必需
    }
  ]
}

和之前的资源一样，client 也可以通过请求来获取可用的 prompt 列表，以及 server 可以通知 client prompt 有更新。

一个实现例子：

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from mcp.server import Server
import mcp.types as types

# Define available prompts
PROMPTS = {
    "git-commit": types.Prompt(
        name="git-commit",
        description="Generate a Git commit message",
        arguments=[
            types.PromptArgument(
                name="changes",
                description="Git diff or description of changes",
                required=True
            )
        ],
    ),
    "explain-code": types.Prompt(
        name="explain-code",
        description="Explain how code works",
        arguments=[
            types.PromptArgument(
                name="code",
                description="Code to explain",
                required=True
            ),
            types.PromptArgument(
                name="language",
                description="Programming language",
                required=False
            )
        ],
    )
}

# Initialize server
app = Server("example-prompts-server")

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return list(PROMPTS.values())

@app.get_prompt()
async def get_prompt(
    name: str, arguments: dict[str, str] | None = None
) -> types.GetPromptResult:
    if name not in PROMPTS:
        raise ValueError(f"Prompt not found: {name}")

    if name == "git-commit":
        changes = arguments.get("changes") if arguments else ""
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"Generate a concise but descriptive commit message "
                        f"for these changes:\n\n{changes}"
                    )
                )
            ]
        )

    if name == "explain-code":
        code = arguments.get("code") if arguments else ""
        language = arguments.get("language", "Unknown") if arguments else "Unknown"
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"Explain how this {language} code works:\n\n{code}"
                    )
                )
            ]
        )

    raise ValueError("Prompt implementation not found")

上述例子定义了两个 prompt：用于生成 commit message 的 git-commit 和用于解释代码的 explain-code ，它们的具体 prompt 内容都在 get_prompt 中定义，而整体定义是在外面。注意在返回 prompt 列表时，是不会返回具体的 prompt 内容的。

整体看下来，这个其实非常像函数的定义，prompt 的 name 实际上就是函数名，prompt 的参数就是函数的参数，而函数体实际上就是把参数带进去，生成我们常见的消息格式，role user content 这一套，最终返回该消息。然后在 get_prompt 中，根据不同的 prompt name 返回不同的 prompt 消息。

所以这样来看，用户是不是就不能更改某个 prompt 的具体内容了？虽然文档上说 prompt 是 user-controlled，但是这里的 control 应该只是选择哪一个 prompt，而不是修改。

Sampling

一听到这个词，自然而然想到是 llm 中的采样，事实也是这样的，只是通过 client 实现的采样。根据文档，sampling 可以让 server 通过 client 去请求 llm，流程如下：

  1. server 向 client 发送一个 sampling/createMessage 请求；
  2. client 审核请求并可修改它；
  3. client 从 LLM 中采样，即执行实际的 llm 请求；
  4. client 审核 completion；
  5. client 将结果返回给 server。

server 向 client 发送的请求格式也基本上是遵循了我们常见的 openai 格式：

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
{
  messages: [
    {
      role: "user" | "assistant",
      content: {
        type: "text" | "image",

        // For text:
        text?: string,

        // For images:
        data?: string,             // base64 encoded
        mimeType?: string
      }
    }
  ],
  modelPreferences?: {
    hints?: [{
      name?: string                // Suggested model name/family
    }],
    costPriority?: number,         // 0-1, importance of minimizing cost
    speedPriority?: number,        // 0-1, importance of low latency
    intelligencePriority?: number  // 0-1, importance of capabilities
  },
  systemPrompt?: string,
  includeContext?: "none" | "thisServer" | "allServers",
  temperature?: number,
  maxTokens: number,
  stopSequences?: string[],
  metadata?: Record<string, unknown>
}

这些字段基本都和使用 openai 请求时差不多，但有一个不同的是，modelPreferences.hints 中的 name ，这个字段实际上填的是可以匹配完整或部分模型名称的字符串，如 “claude-3” 和 “sonnet”。结合下面的几个字段，client 会自动选择指定的模型，比如最省钱、最快或者最智能。

一个请求的实际例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
{
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "What files are in the current directory?"
        }
      }
    ],
    "systemPrompt": "You are a helpful file system assistant.",
    "includeContext": "thisServer",
    "maxTokens": 100
  }
}

响应的格式也和使用 openai 时差不多：

1
2
3
4
5
6
7
8
9
10
11
{
  model: string,  // Name of the model used
  stopReason?: "endTurn" | "stopSequence" | "maxTokens" | string,
  role: "user" | "assistant",
  content: {
    type: "text" | "image",
    text?: string,
    data?: string,
    mimeType?: string
  }
}

Roots

Roots 指的是 workspace 的目录，比如根目录、base url 等。这个功能不是很常用，目前似乎只有 gc 支持，其他都不支持，暂且就不做过多介绍了，我也没有什么使用经验。

END