深入理解MCP：AI 应用与外部世界的桥梁

姘轻拎 · 3 小时前

在构建由大型语言模型（LLM）驱动的智能应用时，如何让 AI 不仅仅停留在文本生成，而是能够与外部世界的数据和工具进行交互，是一个核心挑战。Anthropic 推出的 Model Context Protocol (MCP) 正是为了解决这一问题，它提供了一个标准化的框架，让 AI 应用能够安全、高效地获取上下文信息并调用外部功能。

本文将深入探讨 MCP 的核心概念、三大组件、完整实现、使用流程与传统插件机制的区别。
一、什么是MCP？

MCP (Model Context Protocol) 是一个开源标准和框架，由Anthropic 在2024年11月份提出，旨在连接 AI 应用程序与外部系统。它为 AI 助手提供了一种标准化的方式，使其能够无缝地与外部数据源（如内容管理系统、数据库、企业应用程序等）和各种工具进行集成。
简单来说，MCP 使得 LLM 应用程序能够：

获取实时或领域特定的上下文信息，超越其训练数据的限制。
执行外部操作，例如搜索网页、查询数据库、发送邮件等。

它解决了 AI 助手在与外部世界交互时，如何安全、高效、标准化地获取和利用上下文信息的问题。
二、 MCP 的三大核心组件

MCP 架构由三个关键角色组成，它们协同工作，共同实现了 AI 应用与外部世界的连接：
1. Host (AI 应用本体)

角色： 承载 AI 核心逻辑的应用程序，例如 Cursor、Claude Desktop、Dify、Gptbots等或者您自己开发的 AI应用。
核心功能：
- 管理用户界面 (UI) 和对话历史： 提供用户交互界面，并维护与用户的对话记录。
- 调用 LLM： 负责与底层的大型语言模型（如 OpenAI 的 GPT 系列、Anthropic 的 Claude 系列等）进行交互，发送提示并接收响应。
- 挂载 MCP Client： 在其内部集成 MCP Client，将 MCP Server 暴露的工具映射成 LLM 可以理解和调用的 tools (Function Calling)。
- 处理 tool_calls： 当 LLM 决定调用某个工具时，Host 会将模型生成的 tool_calls 请求，转发给 MCP Client，进而触发对 MCP Server 的实际调用。

2. Client (MCP Client，Host 内的一层运行时)

角色： 位于 Host 内部的运行时层，负责实现 MCP 协议并管理与 MCP Server 的连接。
核心功能：
- 协议实现者 + 进程/连接管理者：
  - 本地 stdio 模式： 负责启动本地的 MCP Server 进程（例如通过 uvx mcp-server-time 或 python mcp_server.py 命令），并通过标准输入/输出 (stdin/stdout) 使用 MCP JSON-RPC 协议进行通信。
  - 示例：获取当前时间的 MCP Server
    比如使用 https://mcp.so 提供的一个获取当前时间的 mcp-server，其配置如下：
    {
      "mcpServers": {
    "time": {
       "command": "uvx",
       "args": [
          "mcp-server-time",
          "--local-timezone=America/New_York"
       ]
    }
      }
    }
    复制代码
    执行原理：
    - 先判断命令 Host 会从 PyPI 上下载 mcp-server-time 这个包。
      
      这块需要强调下协议本身不限制 command 是什么，Host 就是 spawn(command, args) 开子进程；所以你可以内置任意多种：python、uvx、node、npx、docker、自家 CLI……都行，但是host端必须存在对应工具，比如mcp是docker命令执行一个镜像，那host端必须安装了docker，否则启动进程会报错。
    - 找到它在 pyproject.toml 里声明的 console_scripts，例如
      [project.scripts]
      mcp-server-time = "mcp-server-time.server:main"
    - Host 的 MCP Client 会启动一个新进程，执行 my_mcp_demo.server:main()。
    - 在这个 main() 函数内部，会调用 server.run_stdio()，从而使这个新进程成为一个 MCP Server。
    - 这个 mcp-server-time 服务内部都是基于标准 MCP 协议开发的，例如包含 list_tools、call_tool 等方法。
  - 远程 HTTP/SSE 模式： 建立与远程 MCP Server 的 HTTP(S) + SSE（Server-Sent Events）长连接，将 MCP JSON-RPC 消息封装在事件流中发送。

3. Server (MCP Server)

角色： 真正定义“有哪些工具”以及如何执行这些工具的一方。
核心功能：
- 实现 MCP 协议规定的方法： 响应 initialize、tools/list、tools/call 等协议方法。
- 内部注册工具： 注册具体的工具函数，例如 get_current_time (获取当前时间)、web_search (网页搜索) 等，并为每个工具提供其参数的 JSON Schema 定义。
- 返回工具元数据： 当收到 tools/list 请求时，返回所有注册工具的元数据（包括工具名 name、描述 description 和参数 parameters）。
- 执行工具： 当收到 tools/call 请求时，根据工具名和参数执行相应的内部工具，并将执行结果返回。
  外部工具，工具方内部已经实现server了我们直接用即可，内部工具需要我们自己去写server。

三、 MCP 的案例完整实现（可直接使用）

安装完包直接启动mcp_host即可。
项目地址gitee:mcp-test

项目目录

项目依赖

[project]
name = "mcp-test"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"dotenv>=0.9.9",
"openai>=1.0.0",
"mcp>=0.1.0",
]

复制代码

mcp_client.py

import asyncio
from typing import Any, Dict, List
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
class SDKMCPClient:
"""
当前实现仅支持 stdio 模式（command + args 启动本地进程），
如果要支持 URL/HTTP 形式，可以在此基础上再用 SDK 的 HTTP Transport 拓展。
"""
def __init__(self, command: str, args: List[str]) -> None:
self._command = command
self._args = args
async def _run_with_session(self, coro):
server_params = StdioServerParameters(command=self._command, args=self._args)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
return await coro(session)
def list_tools(self) -> List[Dict[str, Any]]:
async def _list(session: ClientSession):
tools = await session.list_tools()
# SDK 返回的是 Tool 对象列表，这里简单转成 dict 方便和方案二统一
tools_meta: List[Dict[str, Any]] = []
for t in tools.tools:
tools_meta.append(
{
"name": t.name,
"description": t.description or "",
"parameters": t.inputSchema or {}, # type: ignore[attr-defined]
}
)
return tools_meta
return asyncio.run(self._run_with_session(_list))
def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
async def _call(session: ClientSession):
result = await session.call_tool(name, arguments)
# result.content 里通常是 TextContent / JsonContent，这里简单返回原始结构
return [c.model_dump() for c in result.content]
return asyncio.run(self._run_with_session(_call))
def close(self) -> None:
# SDK 客户端使用 async context，每次调用都会创建/关闭连接，这里无需额外清理
return None
__all__ = ["SDKMCPClient"]

复制代码

mcp_host.py

import json
import os
import sys
from typing import Any, Dict, List, Tuple
from dotenv import load_dotenv
from openai import OpenAI
from mcp_sdk_client import SDKMCPClient
def build_openai_tools_from_mcp(
server_name: str, tools_meta: List[Dict[str, Any]]
) -> Tuple[List[Dict[str, Any]], Dict[str, str]]:
"""
把某个 MCP Server 返回的工具描述，转换成 OpenAI Chat Completions 的 tools 结构。
同时返回 tool_name -> server_name 的映射。
"""
tools_for_openai: List[Dict[str, Any]] = []
tool_to_server: Dict[str, str] = {}
for t in tools_meta:
name = t["name"]
tool_to_server[name] = server_name
tools_for_openai.append(
{
"type": "function",
"function": {
"name": name,
"description": t.get("description", ""),
"parameters": t.get(
"parameters",
{"type": "object", "properties": {}, "required": []},
),
},
}
)
return tools_for_openai, tool_to_server
def load_mcp_servers_from_files() -> Dict[str, Dict[str, Any]]:
"""
从两个地方聚合 MCP Server 配置：
1. mcp_multi.json （可选，需要维护）
2. 自定义mcp工具
只保留带 command/args 的本地进程型 MCP server。
"""
servers: Dict[str, Dict[str, Any]] = {}
def merge_from_mcpservers_obj(obj: Any, src: str) -> None:
"""从一个可能包含 mcpServers 字段的对象里合并配置。"""
if not isinstance(obj, dict):
return
servers_cfg = obj.get("mcpServers") or {}
if not isinstance(servers_cfg, dict):
print(f"[Console-Host] 跳过 {src}：mcpServers 必须是对象。")
return
for name, cfg in servers_cfg.items():
if not isinstance(cfg, dict):
continue
command = cfg.get("command")
args = cfg.get("args", [])
if not command or not isinstance(args, list):
continue
servers[name] = {
"command": str(command),
"args": [str(a) for a in args],
}
def load_from_path(path: str) -> None:
"""
从JSON 文件里合并 MCP Server 配置。[ {"mcpServers": {...}}, {"mcpServers": {...}}, ... ]
"""
if not os.path.exists(path):
return
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
for idx, item in enumerate(data):
merge_from_mcpservers_obj(item, f"{path}[{idx}]")
except Exception as exc:
print(f"[Console-Host] 读取 {path} 失败: {exc}")
# 1) 本地项目内的配置
load_from_path(os.path.join(os.path.dirname(__file__), "mcp_multi.json"))
# 2) 内置两个 demo MCP Server（始终可用）
base_dir = os.path.dirname(__file__)
servers.setdefault(
"demo-math",
{
"command": sys.executable,
"args": [os.path.join(base_dir, "demo_math_server.py")],
},
)
servers.setdefault(
"demo-text",
{
"command": sys.executable,
"args": [os.path.join(base_dir, "demo_text_server.py")],
},
)
return servers
def main() -> None:
"""
- 从 ./mcp_multi.json 聚合所有本地自定义 MCP server
- 对每个 server 通过 SDKMCPClient.list_tools() 拉取工具
- 聚合所有工具为一个大 tools 列表
- 控制台循环读入用户输入，让 OpenAI 模型按需调用这些工具
"""
load_dotenv("dev.env")
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
print("请先设置环境变量 OPENAI_API_KEY，例如：")
print(" set OPENAI_API_KEY=你的key （PowerShell: $env:OPENAI_API_KEY="..."）")
return
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
oa_client = OpenAI(api_key=api_key)
servers = load_mcp_servers_from_files()
if not servers:
print("[Console-Host] 没有找到任何 MCP Server 配置。")
print("你可以在 mcp_multi.json 中配置 mcpServers。")
return
print(f"[Console-Host] 已加载 MCP Server 配置: {list(servers.keys())}")
# 启动所有 MCP server，聚合 tools
clients: Dict[str, SDKMCPClient] = {}
server_tools: Dict[str, List[Dict[str, Any]]] = {}
tools_for_openai: List[Dict[str, Any]] = []
tool_to_server: Dict[str, str] = {}
for name, cfg in servers.items():
command = cfg["command"]
args = cfg["args"]
print(f"[Console-Host] 连接 MCP Server '{name}'，命令: {[command] + args}")
try:
client = SDKMCPClient(command=command, args=args) # 启动server进程
tools_meta = client.list_tools()
except Exception as exc: # noqa: BLE001
print(f"[Console-Host] 连接 '{name}' 失败: {exc}")
continue
clients[name] = client
server_tools[name] = tools_meta
part_tools, mapping = build_openai_tools_from_mcp(name, tools_meta)
tools_for_openai.extend(part_tools)
tool_to_server.update(mapping)
print(f"[Console-Host] Server '{name}' 工具: {[t['name'] for t in tools_meta]}")
if not clients:
print("[Console-Host] 所有 MCP Server 都连接失败，退出。")
return
print(f"[Console-Host] 最终聚合工具数: {len(tools_for_openai)}")
print("现在可以开始对话了，输入 exit / quit 结束。")
messages: List[Dict[str, Any]] = []
while True:
try:
user_input = input("你：").strip()
except (EOFError, KeyboardInterrupt):
print()
break
if not user_input:
continue
if user_input.lower() in {"exit", "quit"}:
break
messages.append({"role": "user", "content": user_input})
try:
response = oa_client.chat.completions.create(
model=model,
messages=messages,
tools=tools_for_openai,
tool_choice="auto",
)
except Exception as exc: # noqa: BLE001
print(f"[Console-Host] 调用 OpenAI 出错: {exc}")
messages = []
continue
msg = response.choices[0].message
tool_calls = msg.tool_calls or []
if tool_calls:
print("[Console-Host] 模型决定调用 MCP 工具:")
assistant_tool_calls_payload: List[Dict[str, Any]] = []
for tc in tool_calls:
print(
f" - 工具名: {tc.function.name}, "
f"arguments: {tc.function.arguments}"
)
assistant_tool_calls_payload.append(
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
)
messages.append(
{
"role": "assistant",
"tool_calls": assistant_tool_calls_payload,
"content": None,
}
)
# 路由到对应的 MCP Server
for tc in tool_calls:
args = json.loads(tc.function.arguments or "{}")
tool_name = tc.function.name
server_name = tool_to_server.get(tool_name)
client = clients.get(server_name) if server_name else None
if not client:
print(
f"[Console-Host] 找不到处理工具 '{tool_name}' 的 MCP Server（映射缺失）。"
)
continue
print(
f"[Console-Host] 调用 MCP Server '{server_name}' 工具 '{tool_name}'，参数: {args}"
)
try:
tool_result = client.call_tool(tool_name, args)
except Exception as exc: # noqa: BLE001
print(
f"[Console-Host] 调用 MCP Server '{server_name}' 工具 '{tool_name}' 失败: {exc}"
)
continue
print(
f"[Console-Host] 工具 '{tool_name}' 返回结果: {tool_result}"
)
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(tool_result, ensure_ascii=False),
}
)
try:
response = oa_client.chat.completions.create(
model=model,
messages=messages,
)
except Exception as exc: # noqa: BLE001
print(f"[Console-Host] 第二次调用 OpenAI 出错: {exc}")
continue
final_msg = response.choices[0].message
answer = final_msg.content or ""
else:
answer = msg.content or ""
print(f"助手：{answer}")
messages.append({"role": "assistant", "content": answer})
if __name__ == "__main__":
main()

复制代码

mcp_multi.json

[
{
"mcpServers": {
"context7": {
"command": "npx",
"args": [
"-y",
"@upstash/context7-mcp",
"--api-key",
"ctx7sk-9db997d9-9eeb-4c4b-9b0d-7b4f2656a8db"
]
}
}
},
{
"mcpServers": {
"time": {
"command": "uvx",
"args": [
"mcp-server-time",
"--local-timezone=America/New_York"
]
}
}
},
{
"mcpServers": {
"howtocook-mcp": {
"command": "npx",
"args": [
"-y",
"howtocook-mcp"
]
}
}
}
]

复制代码

可以自行去mcp.so官网找一些可以启本地进程的mcp

demo_math_server.py

import asyncio
import random
from typing import Annotated
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo-math")
@mcp.tool()
async def add(
a: Annotated[float, "第一个数字"],
b: Annotated[float, "第二个数字"],
) -> Annotated[float, "a + b 的结果"]:
"""把两个数字相加。"""
await asyncio.sleep(0) # 协程占位
return a + b
@mcp.tool()
async def random_int(
minimum: Annotated[int, "最小整数（包含）"],
maximum: Annotated[int, "最大整数（包含）"],
) -> Annotated[int, "生成的随机整数"]:
"""在给定区间内生成一个随机整数。"""
await asyncio.sleep(0)
if minimum > maximum:
minimum, maximum = maximum, minimum
return random.randint(minimum, maximum)
def main() -> None:
"""
demo 数学 MCP Server：
- 工具 add(a, b)
- 工具 random_int(minimum, maximum)
通过 stdio 运行，满足 MCP 协议，可被 SDK 客户端连接。
"""
mcp.run()
if __name__ == "__main__":
main()

复制代码

demo_text_server.py

import asyncio
from typing import Annotated
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo-text")
@mcp.tool()
async def upper(
text: Annotated[str, "要转换为大写的字符串"],
) -> Annotated[str, "大写结果"]:
"""把字符串转成大写。"""
await asyncio.sleep(0)
return text.upper()
@mcp.tool()
async def word_count(
text: Annotated[str, "要统计的文本"],
) -> Annotated[int, "按空格分词后的词数"]:
"""统计一句话里有多少个词（按空格粗略切分）。"""
await asyncio.sleep(0)
return len(text.split())
def main() -> None:
"""
demo 文本 MCP Server：
- 工具 upper(text)
- 工具 word_count(text)
通过 stdio 运行，满足 MCP 协议，可被 SDK 客户端连接。
"""
mcp.run()
if __name__ == "__main__":
main()

复制代码

随便写了两个内置工具，可以自行修改测试

dev.env

OPENAI_API_KEY=sk-proj-xxxx
OPENAI_MODEL=gpt-4o

复制代码

四、典型的 MCP 使用流程

用户配置 MCP Server 信息： 用户在 AI 应用（Host）的配置中（例如一个 mcp.json 文件）定义了有哪些 MCP Server，以及如何连接它们（是本地进程 stdio 模式还是远程 URL+SSE 模式）。
Host 建立连接并获取工具列表：
- Host 读取配置后，通过其内部的 MCP Client 建立与 MCP Server 的连接（启动本地进程或连接远程 URL）。
- Client 按照 MCP 协议向 Server 发送 tools/list 请求。
- MCP Server 响应请求，返回其内部注册的所有工具的列表，包括每个工具的名称、描述和参数的 JSON Schema。
Host 映射工具为 LLM 可调用格式： Host 接收到 MCP Server 返回的工具列表后，将这些工具的元数据转换成 LLM（如 OpenAI 的 Function Calling 机制）可以理解和调用的 tools 格式。这相当于“将 MCP 世界的工具安装进模型的插件系统”。
用户对话，LLM 判断是否需要工具：
- 用户与 AI 应用进行对话。
- Host 将用户输入和已映射的工具定义一同发送给 LLM（通过 chat.completions API 调用，并带上 tools 参数）。
- LLM 根据对话内容和工具定义，判断是否需要调用某个工具来完成任务。如果需要，模型会返回一个 tool_calls 响应，指明它“想用 MCP 里的某个工具”以及相应的参数。
Host 用 MCP Client 调 tools/call 真正执行工具：
- 当 Host 收到 LLM 返回的 tool_calls 时，它会通过 MCP Client 再次向 MCP Server 发送 tools/call 请求，真正执行模型指定的工具，并传入模型生成的参数。
- MCP Server 执行相应的内部工具，并将执行结果返回给 MCP Client。
Host 把结果塞回对话，再让 LLM 给最终回答：
- Host 收到工具执行结果后，将这个结果作为新的上下文信息，再次塞回给 LLM（作为 tool_outputs）。
- LLM 结合之前的对话历史和工具执行结果，生成最终的回答并返回给用户。

举例来说：
用户在 Cursor（Host）中配置了一个新的 MCP Server。Cursor 内置的 MCP Client 会立即启动这个 MCP 进程（如果是本地模式）或建立远程连接。Host 随后通过 Client 拉取 MCP Server 暴露的所有工具，并将它们转换为 OpenAI 的 tools 格式。当用户提问时，模型可能会决定调用一个 MCP 工具（例如 web_search），Host 通过 tools/call 将请求转发给 MCP Server。Server 执行搜索并返回结果，Host 再将搜索结果提供给模型，让模型生成最终的最终回答。
五、 MCP 与传统插件机制的区别

mcp实际就是提供了一种标准化的访问外部数据源的方式，他能做的插件也能做，但是插件与模型厂商挂钩，协议格式各不相同，对接复杂，mcp简化了这种流程，且更加开放标准了别人写好了mcp工具我们直接就能用。

在构建基于大型语言模型（LLM）的应用时，理解 function_call 和 Model Context Protocol (MCP) 这两个概念至关重要。它们虽然都与工具集成相关，但作用的“层级”和解决的问题截然不同。
1. function_call 是“LLM API 级别”的能力

function_call（或类似的工具调用机制，如 Anthropic 的 tool_use）是 LLM 提供商在其 API 中内置的一种能力。它解决的核心问题是：“这个模型如何在一次 API 调用里请求某个函数、传参，并让调用者拿到结果？”
对于开发者来说，使用 function_call 时，你需要自己搞定以下这些“手工工作”：

工具发现： 去哪里找到这些可供 LLM 调用的函数（工具）？多个 server、几十上百个工具，怎么声明、分类、动态启用/禁用？
工具传输和生命周期： 这些函数（工具）是本地运行的、需要通过远程 HTTP API 调用的，还是通过命令行接口（CLI）执行的？连接、心跳、长任务、cancel、错误码处理？
服务厂商差异化处理： 如果有多个工具服务提供商，它们的 API schema/鉴权/错误码可能各不相同，如何统一处理？

因此，在没有 MCP 这样的协议层时，你的 AI 应用（Host）里，实际上做了很多繁琐且定制化的工作：

从 JSON 配置文件或特定的 MCP Server 拉取工具元数据。
把它们“翻译”成 OpenAI 或其他 LLM 平台所要求的 tools（Function Calling）格式。
收到 tool_calls 再自己路由到不同的 server 去执行。

2. MCP 是“工具/Agent 生态层”的协议

Model Context Protocol (MCP) 则是一个更高层级的协议，它旨在解决更宏观的问题：“世界上所有想给 LLM 用的工具/Agent，要用什么统一的方式把自己挂出来，让任何 AI 应用都能方便地发现和使用？”
MCP 协议提供了一套标准化的机制，包括：

外部工具能力标准化（发现 + 传输）
- 工具发现：mcpServers 配置 + tools/list → Host 不需要为每个服务商自定义“列出我有哪些能力”的协议。
- tools/call + JSON‑RPC over stdio/HTTP/SSE → 不同传输方式下，语义和报文结构是一致的。
工具协议处理（鉴权 / 请求 / 响应 / 错误码）
- 各家服务商在自己的 MCP Server 里，把乱七八糟的内部 API（鉴权、数据结构、错误码）统统“翻译”为统一的 MCP 形状；
- Host 只跟 MCP Server 说话，看到的是统一的：
  - initialize / tools/list / tools/call
  - 标准 JSON‑RPC 报文
  - 标准 error.code / error.message / error.data。
多端复用
- 同一个 MCP Server，可以被不同的 AI 应用（Host），如 Cursor、Claude Desktop 或你自己的自定义 Host，直接连接并复用，极大地提高了工具的生态复用性。

对于 AI 应用（Host）来说，有了 MCP 这一层协议之后，其工作变得更加简化和标准化：

通过client按照mcp协议连上 mcp-server；
host连接client获取server的所有工具挂到 llm-function_call 里；
按规范把 tools.call 转发出去就行，不用管每家怎么实现。

可以这么说：在“能不能让模型调工具”这件事上，function_call 理论上都能做到 MCP 能做的事，但两者不在一个层级，MCP 是把一整层东西“标准化 + 外包”了。
总结，function_call 是 LLM 本身具备的“调用函数”的能力，而 MCP 则是一套“如何标准化地组织、发现和调用这些函数”的协议和生态系统。MCP 极大地简化了 Host（AI应用）在工具集成方面的工作，将复杂性下沉到 Client 和 Server 层，从而促进了 AI 工具生态的繁荣和互操作性。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

深入理解MCP：AI 应用与外部世界的桥梁

相关帖子

签约作者

深入理解MCP：AI 应用与外部世界的桥梁

相关帖子

相关推荐

签约作者