MCP（Model Context Protocol，模型上下文协议）是一种开放标准协议，由Anthropic于2024年11月提出并开源。其主要目的是在大型语言模型（LLM）与外部数据源及工具之间建立安全的双向链接。
MCP协议旨在解决AI模型因数据孤岛限制而无法充分发挥潜力的问题，使AI应用能够安全地访问和操作本地及远程数据。此外，MCP通过标准化的服务器实现，支持AI模型与各种资源进行安全交互，使其成为连接万物的接口。
MCP有如下几个优势：
- 将工具注入从编译时转移到运行时，开发者可以在运行时动态管理工具的增删改。
- 可以将工具运行从本地转移到分布式的远程Server，有效减少了本地依赖，轻量化应用。
- 工具得到了有效的抽象，可以跨应用提供安全的数据访问。
MCP 的底层通过 JSON-RPC 2.0 协议实现通信，支持两种传输机制：
- 标准输入输出（本地进程通信）
- Server-Sent Events（SSE over HTTP，用于远程通信）
从官方给出的样例代码中可以看到，本质上是通过MCP协议拿到所有的工具信息后，通过一定的格式化后，拼接到prompt里发送给大模型。
然后根据大模型的输出来决定工具的调用
总结一下LLM通过MCP调用工具的执行流程
1
2
3
4
5
6
7
8
9
10
11
12
13
|
# 给项目创建一个文件夹
uv init mcpdemo
cd mcpdemo
# 创建一个虚拟环境并激活
uv venv
source .venv/bin/activate
# 安装依赖
uv add "mcp[cli]" httpx
# 创建 server 文件
touch server_demo.py
|
这里分别开发了两种协议的demo
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
# -*- coding: utf-8 -*-
import logging
from typing import Any
import asyncio
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
from pydantic import AnyUrl
server = Server("demo")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
"""
提示模版定义
"""
return [
types.Prompt(
name="example-prompt",
description="An example prompt template",
arguments=[
types.PromptArgument(
name="arg1",
description="Example argument",
required=True
)
]
)
]
@server.get_prompt()
async def handle_get_prompt(
name: str,
arguments: dict[str, str] | None
) -> types.GetPromptResult:
"""
提示模板处理
"""
if name != "example-prompt":
raise ValueError(f"Unknown prompt: {name}")
return types.GetPromptResult(
description="Example prompt",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text="Example prompt text"
)
)
]
)
@server.list_resources()
async def list_resources() -> list[types.Resource]:
"""
资源定义
"""
test='test.txt'
return [
types.Resource(
uri=AnyUrl(f"file:///tmp/{test}"),
name=test,
description=f"A sample text resource named {test}",
mimeType="text/plain",
)
# for name in SAMPLE_RESOURCES.keys()
]
@server.read_resource()
async def read_resource(uri: AnyUrl) -> str | bytes:
assert uri.path is not None
with open('/tmp/test.txt', 'r') as file:
data = file.read()
return data
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
工具定义.
每个工具都使用JSON Schema验证指定其参数.
"""
return [
types.Tool(
name="demo-tool",
description="Get data tool for a param",
inputSchema={
"type": "object",
"properties": {
"param": {
"type": "string",
"description": "url",
},
},
"required": ["param"],
},
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[Any]:
logging.info(name)
"""
处理工具调用
"""
if not arguments:
raise ValueError("Missing arguments")
if name == "demo-tool":
param = arguments.get("param")
if not param:
raise ValueError("Missing state parameter")
param = param.upper()
return [
types.TextContent(
type="text",
text=f"text:{param}"
)
]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
from anyio.streams.text import TextStream
# 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="demo-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())
|
server_demo.py内容如下，这里是MCP官方SDK中的一个SSE传输的样例（代码里其实也支持了stdio）
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
import anyio
import click
import httpx
import mcp.types as types
from mcp.server.lowlevel import Server
async def fetch_website(
url: str,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
headers = {
"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
}
async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:
response = await client.get(url)
response.raise_for_status()
return [types.TextContent(type="text", text=response.text)]
@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option(
"--transport",
type=click.Choice(["stdio", "sse"]),
default="stdio",
help="Transport type",
)
def main(port: int, transport: str) -> int:
app = Server("mcp-website-fetcher")
@app.call_tool()
async def fetch_tool(
name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
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"])
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="fetch",
description="Fetches a website and returns its content",
inputSchema={
"type": "object",
"required": ["url"],
"properties": {
"url": {
"type": "string",
"description": "URL to fetch",
}
},
},
)
]
if transport == "sse":
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount, Route
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()
)
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
import uvicorn
uvicorn.run(starlette_app, host="127.0.0.1", port=port)
else:
from mcp.server.stdio import stdio_server
async def arun():
async with stdio_server() as streams:
await app.run(
streams[0], streams[1], app.create_initialization_options()
)
anyio.run(arun)
return 0
if __name__ == '__main__':
main()
|
运行server
1
|
python server_demo.py --port 8800 --transport sse # sse模式
|
官方提供了一个调试工具inspector，可以本地调试Server的功能
1
|
npx -y @modelcontextprotocol/inspector
|
注意这里要防止环境变量跟工作目录带来的问题，最好填写绝对路径。
连接上后可以看到读出了我们设置的test.txt文件内容
获取prompt
工具调用，由于只是模拟，这里直接返回了传入参数的大写。
在左边切换协议，还支持SSE的测试。可以看到成功抓取了我博客的内容。
同样，还可以调用别人写好的MCP工具
1
|
npx -y @modelcontextprotocol/server-filesystem .
|
MCP提供了统一的规范和标准，大家都能去按照这个协议去开发插件，共享劳动成果。
Cline是Vscode的一个插件，支持MCP协议的工具调用，同时内置了插件市场，可以一键安装使用别人写好的MCP工具。
插件市场
在右上角设置里增加我们的MCP Server
1
2
3
4
5
6
7
8
9
10
11
12
|
{
"mcpServers": {
"server_demo": {
"command": "/Users/yzddmr6/PycharmProjects/MCPDemo/venv/bin/python",
"args": [
"/Users/yzddmr6/PycharmProjects/MCPDemo/server.py"
],
"disabled": false,
"autoApprove": []
}
}
}
|
然后再去列举工具
调用MCP工具
Cline可以交互式的调用执行一些命令，并根据输出来反思调整
不过有时候模型并不是非常聪明，这时候就需要人工参与了
还可以在插件市场里调用别人的插件，这里以官方推出的File System为例
但是发现一个问题，Cline目前还并不支持SSE远程调用。
翻了一下issue，官方已经有人在跟进了，但是看起来工作量比较大
那么如何解决呢？有人提到可以用supergateway这个插件做一层中转，把sse转为stdio：https://github.com/supercorp-ai/supergateway
经过测试本地直接运行命令没有问题，但是在Cline上配置后报错，不太清楚是什么原因
1
|
npx -y supergateway --sse http://127.0.0.1:8800/sse
|
Cursor同时支持命令行与sse两种模式，遥遥领先
可以看到控制台有连接记录了
但非Pro用户使用非常不稳定，没充钱就是这样子的。
本文我们探讨了MCP的背景、运行原理以及其在开发AI应用中的实际应用。
MCP的出现正好解决了我最近工作中遇到的一些问题：原来的工具和模型是耦合的，而模型的迭代速度又非常快，如果想要做迁移就要重新写一份代码。解耦之后，可以更关注业务逻辑，减少重复劳动，同时也可以共享互联网上别人写好的工具和资源。
现在越来越多的组件和公司开始兼容MCP协议，甚至OpenAI的Agent SDK最近也已经支持了MCP。要知道，推出MCP协议的的是他的头号竞争对手Anthropic。
目前mcp.so收录了近4600+ MCP Server，包括微软、高德、百度地图等都已经推出了自己的MCP Server。smithery.ai也收录了近3000个。
个人认为，MCP的价值核心在于支持远程的调用。官方前两天也给MCP远程调用增加了鉴权的支持，采用OAuth2.1标准。
https://spec.modelcontextprotocol.io/specification/2025-03-26/changelog/
对安全行业来说，MCP的普及可能会带来新的攻击面，例如未授权，越权等。毕竟，如果是远程部署的MCP Server，我连上了你 = 我能访问你的数据库，读取你的文件，甚至执行任意的命令。
可以预见的是，MCP在接下来相当长的一段时间内会成为行业的标准和共识，越来越多的开发者和公司会加入到这个开放生态系统中，共享开发的工具和资源，加速AI应用的创新和发展。
