什么是MCP

MCP 起源于 2024 年 11 月 25 日 Anthropic发布的文章：Introducing the Model Context Protocol。

MCP （Model Context Protocol，模型上下文协议）定义了应用程序和 AI 模型之间交换上下文信息的方式。这使得开发者能够以一致的方式将各种数据源、工具和功能连接到 AI 模型（一个中间协议层），就像 USB-C 让不同设备能够通过相同的接口连接一样。MCP 的目标是创建一个通用标准，使 AI 应用程序的开发和集成变得更加简单和统一。

想象一下没有 MCP 之前我们会怎么做？我们可能会人工从数据库中筛选或者使用工具检索可能需要的信息，手动的粘贴到 prompt 中。随着我们要解决的问题越来越复杂，手工把信息引入到 prompt 中会变得越来越困难。

为了克服手工 prompt 的局限性，许多 LLM 平台（如 OpenAI、Google）引入了 function call 功能。这一机制允许模型在需要时调用预定义的函数来获取数据或执行操作，显著提升了自动化水平。

MCP的结构

  • MCP Hosts: 像 Claude Desktop、IDEs 或 AI 工具这样的程序，它们希望通过 MCP 访问资源
  • MCP Clients: 维护与服务器 1:1 连接的协议客户端
  • MCP Servers: 轻量级程序，通过标准化的 Model Context Protocol 暴露特定功能
  • Local Resources: 你的计算机资源（数据库、文件、服务），MCP 服务器可以安全地访问这些资源
  • Remote Resources: 通过互联网可用的资源（例如，通过 APIs），MCP 服务器可以连接到这些资源

创建MCP Server

使用uv创建应该环境

1
uvx create-mcp-server --path 路径

然后按照它的提示一步步创建即可，并添加对它的依赖

1
uv add httpx python-dotenv

编写MCP

进入项目的src/项目同名文件夹，可以看到有以下创建好的两个python文件

1
2
3
4
Mode                 LastWriteTime         Length Name
----                 -------------         ------ ----
-a----         2025/3/11     19:15           5076 server.py
-a----         2025/3/11     19:15            220 __init__.py

其中我们进去会发现它已经为我们写好一个可以使用的例子了，但是我们还是需要了解其中内容的结构的。

首先是server部分，这是mcp能实现功能的关键部分

资源部分

当你想让ai访问本地/线上资源的时候，ai怎么知道有哪些资源可以访问？又该如何阅读？资源部分就是为了解决这个问题而诞生的，我们只需要为ai声明阅读的uri并实现阅读的方法，就能进行资源使用。

资源声明结构

这是资源声明的格式

1
2
3
4
5
6
{
  uri: string;           // 资源的唯一标识符
  name: string;          // 人类可读的名称
  description?: string;  // 可选描述
  mimeType?: string;     // 可选 MIME 类型
}
1
2
3
4
5
6
{
  uriTemplate: string;   // 遵循 RFC 6570 的 URI 模板
  name: string;          // 此类型的人类可读名称
  description?: string;  // 可选描述
  mimeType?: string;     // 所有匹配资源的可选 MIME 类型
}

格式的 URI 进行标识：

1
[protocol]://[host]/[path]

例如：

  • file:///home/user/documents/report.pdf
  • postgres://database/customers/schema
  • screen://localhost/display1(获取屏幕信息)

创建资源标识可以使得资源被ai发现阅览，而这一过程通过以下两个函数传递

list_resources()

这个函数的作用是将资源以列表的列表传递给大模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        # 直接资源，这里给出的uri是确定的，ai只需要对uri进行选择
        types.Resource(
            uri="uri路径",
            name="资源名称",
            mimeType="资源类型"
        )
        # 资源模板，这里的uri是ai根据RFC 6570 的 URI 模板进行自动输入，详情 https://rfc2cn.com/rfc6570.html
        types.Resource(
            uriTemplate="uri路径",
            name="资源名称",
            mimeType="资源类型"
        )
    ]

read_resource()

处理并阅读其中的资源，将其以自定义的形式返回的大模型

1
2
3
4
5
6
7
8
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    # 根据ai要访问的资源路径进行处理，这里的uri则是由ai生成
    if str(uri) == "file:///logs/app.log":   # 此处根据上面资源文件或是资源模板的uri进行判断来选择执行不同的资源处理
        log_contents = await read_log_file() # 任意函数处理文件，最终以string的格式返回给ai
        return log_contents

    raise ValueError("资源未找到")

工具部分

工具结构

1
2
3
4
5
6
7
8
{
  name: string;          // 工具的唯一标识符
  description?: string;  // 人类可读的描述
  inputSchema: {         // 工具参数的 JSON Schema
    type: "object",
    properties: { ... }  // 工具特定的参数
  }
}

list_tools实现工具列表

这个函数的作用是告诉ai可以使用的工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="工具函数名",
            description="工具的描述",
            inputSchema={
                "type": "object",
                "properties": {
                    "变量名1": {
                        "type": "string",
                        "description":"变量信息描述"
                    },
                    "变量名2": {"type": "string"},
                },
                "required": ["变量名1", "变量名2"],
            },
        )
    ]

handle_call_tool

这个函数将根据ai的返回进行分析处理并返回给ai

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
@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    Handle tool execution requests.
    Tools can modify server state and notify clients of changes.
    """
    if name != "函数名": 		#判断当前ai调用的函数
        raise ValueError(f"Unknown tool: {name}")

    if not arguments:			#判断是否输入进参数（参数以dict的格式传入）
        raise ValueError("Missing arguments")

    变量名1 = arguments.get("变量名1")
    变量名2 = arguments.get("变量名2")

    # 可以添加调用其他的处理函数

    return [
        types.TextContent(
            type="text",
            text=f"返回给ai的内容",
        )
    ]

提示部分

我们给了大模型工具的可以访问的资源，但是和人一样，面临很多新的名词可能无法进行理解，如何使用就成了新的问题，提示部分就好比一本说明书，对那些名词进行说明并告知大模型应该怎样进行运行。

这一过程分为两个部分：

  1. 发现提示
  2. 使用提示

提示部分结构

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

定义提示

那我们如何在python中写这个功能呢？首先需要根据提示结构一个提示的静态变量，用来说明有哪些提示

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
PROMPTS = {
    "git-commit": types.Prompt(
        name="git-commit",
        description="生成 Git 提交消息",
        arguments=[
            types.PromptArgument(
                name="changes",
                description="Git diff 或更改描述",
                required=True
            )
        ],
    ),
    "explain-code": types.Prompt(
        name="explain-code",
        description="解释代码如何工作",
        arguments=[
            types.PromptArgument(
                name="code",
                description="要解释的代码",
                required=True
            ),
            types.PromptArgument(
                name="language",
                description="编程语言",
                required=False
            )
        ],
    )
}

传递提示列表

生成提示列表则是为了让ai知晓存在那些提示

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

大模型会对server端进行一次询问

1
2
3
4
// 请求
{
  method: "prompts/list"
}

此时list_prompts则会返回我们的提示信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 响应
{
  prompts: [
    {
      name: "analyze-code",
      description: "分析代码以寻找潜在的改进",
      arguments: [
        {
          name: "language",
          description: "编程语言",
          required: true
        }
      ]
    }
  ]
}

get_prompt解析提示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@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"未找到提示：{name}")

    if name == "提示名":
        changes = arguments.get("参数名称") if arguments else ""
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text", # 类型
                        text=f"为这些更改生成简洁但描述性的提交消息：\n\n{changes}" # 返回的信息
                    )
                )
            ]
        )


    raise ValueError("未找到提示实现")

启动函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
async def main():
    # Run the server using stdin/stdout streams
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="项目名称",
                server_version="版本",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),

放入mcp

在其设置目录中放入

1
2
3
4
5
6
7
8
9
10
11
12
13
{
    "mcpServers": {
        "weather": {
            "command": "uv",
            "args": [
                "--directory",
                "项目路径",
                "run",
                "项目名称"
            ]
        }
    }
}