MCP 学习笔记

---

目录

• 1 MCP入门
• 2 MCP服务器
• 3 MCP客户端
• 4 MCP进阶使用
• 5 Cursor接入MCP

---

5.1 MCP入门

MCP 介绍

什么是 MCP

MCP是一种开放的技术协议，旨在标准化大型语言模型（LLM）与外部工具和服务的交互方式。你可以把MCP理解成是一个AI世界的通用翻译官，让AI模型能够与各种各样的外部工具"对话"。

为什么需要 MCP

在MCP出现之前，AI工具调用面临两大痛点：

• 接口碎片化：每个LLM使用不同的指令格式，每个工具API也有独特的数据结构，开发者需要为每个组合编写定制化连接代码；
• 开发低效：这种"一对一翻译"模式成本高昂且难以扩展，就像为每个外国客户雇用专属翻译。

模型与工具深度绑定，无法构建统一生态系统，大大增加了迁移成本。

而MCP则采用了一种通用语言格式（JSON-RPC），一次学习就能与所有支持这种协议的工具进行交流。一个通用翻译器，不管什么LLM，用上它就能使用工具/数据了。

MCP的特点

• 协议标准化 – 统一 JSON-RPC 2.0 消息格式，任何兼容 MCP 的主机（Claude Desktop、Cursor 等）都能即插即用。
• 本地优先 – 敏感数据留在本地，API 密钥不再暴露给模型提供商。
• 动态扩展 – 新增工具只需启动一个新的 MCP Server，无需改模型或主程序。
• 安全隔离 – 双层授权 + 最小权限 + 端到端加密。

MCP典型使用场景

场景	举例	MCP 的作用
个人 AI 助手	让 Claude Desktop 直接查本地日历、写文件、发邮件	利用 MCP Server 暴露文件系统、邮件接口
企业自动化	在 IDE 里通过自然语言生成 SQL、操作 ERP、CRM	企业内网部署 MCP Server，连接内部系统
数据分析	实时拉取多源数据（PostgreSQL、Google Sheet、第三方 API）做报表	通过 Server 把各种数据源统一暴露给模型
社区生态	一键安装 @wopal/mcp-server-hotnews 获取热门新闻	共享 Server 市场，像装 VS Code 插件一样简单

MCP vs Function Calling vs Tool

维度	MCP	Function Calling	Tool
本质	开放协议/标准	模型自身能力	泛指可被调用的外部函数
架构	Client–Server（主机-服务器）	模型直接生成调用指令	无统一架构
耦合度	低：一次开发，多模型通用	高：需针对模型微调 prompt / schema	无标准，各自为政
扩展方式	启动新 MCP Server 即可	需改 prompt、重训或改代码	每换环境重写
安全模型	本地运行、双层授权	密钥/数据通常送云端	取决于实现
类比	USB-C 通用接口	各品牌私有充电线	单个转接头

Function Call是大型语言模型（LLM）与外部工具或API交互的核心机制。它是大模型的一个基础能力，就是识别什么时候要工具，可能需要啥类型的工具的能力。

而MCP则是工具分类的箱子。因此MCP不是要取代Function Call，而是要在Function Call基础上，联合Agent一起去完成复杂任务。

如果把整个工具调用的流程剖析开来，实际是"Function Call + Agent + MCP系统的组合"。用一句话说清楚：大模型通过Function Call表达"我要调用什么工具"，Agent遵循指令执行工具的调用，而MCP则是提供了一种统一的工具调用规范。

架构原理

工作原理

MCP采用CS架构（客户端-服务器），MCP的技术架构可以简单理解为一个由三个核心部分组成的系统：MCP Host、MCP Client和MCP Server。

MCP主机（MCP Hosts）

MCP主机是发起请求的AI应用程序，它们希望通过MCP协议访问外部数据或工具。例如：Claude Desktop、AI驱动的IDE（如Cursor）、其他AI应用（如聊天机器人、自动化助手等）

MCP客户端（MCP Clients）

MCP客户端是MCP主机和MCP服务器之间的桥梁，与服务器一对一进行连接，负责与服务器进行通信，执行数据请求和工具调用。它的主要功能包括：

• 从MCP服务器获取可用的工具列表。
• 将用户的查询和工具描述一起发送给大模型。
• 接收大模型的决策，判断是否需要使用工具。
• 通过MCP服务器调用相应的工具，并获取返回结果。
• 将结果反馈给大模型，由大模型生成最终的自然语言响应。

MCP服务器（MCP Servers）

MCP服务器是整个架构的核心，它实现了MCP协议，并提供各种功能来支持AI应用。它主要负责：

• 资源：提供可被读取的数据，如本地文件、API响应、数据库等。
• 工具：提供可以被大模型调用的函数或操作。
• 提示词：提供预定义的提示词模板，帮助用户完成特定任务。

每个MCP服务器通常专注于特定的任务，例如：读取和写入浏览器数据、访问本地文件系统、操作Git仓库、连接远程API等等。

本地数据源（Local Data Sources）

MCP服务器可以访问计算机上的本地资源，例如：本地文件（如PDF、Word文档、代码文件）、本地数据库（如SQLite、PostgreSQL）、其他本地应用的数据等等。本地数据源的特点是数据不会上传到远端，确保数据安全性。

远程服务（Remote Services）

MCP服务器也可以连接到远程资源，例如：在线API、企业内部系统、其他基于云端的数据服务等等。远程服务通常通过API访问，并由MCP服务器进行管理，来确保访问权限的控制。

通信原理

通信协议

MCP采用JSON-RPC作为底层的通信协议。JSON-RPC是一种基于JSON的轻量级远程调用协议，相较于HTTP来说它更加简洁、高效、容易处理。

属性	HTTP	JSON-RPC
本质	应用层协议（Web核心协议）	轻量级RPC协议（基于JSON格式）
数据格式	支持JSON/XML/二进制等多种格式	强制JSON格式，结构更简洁
协议功能	包含缓存/认证/状态码等完整功能	仅定义RPC调用规范（无底层逻辑）
通信模式	无状态，支持GET/POST等多方法	无状态，基于method字段调用
适用场景	Web API、浏览器交互、复杂业务	微服务内部调用、物联网等轻量场景
典型应用	RESTful接口、网页加载	服务间函数调用、嵌入式设备通信

通信方式

MCP基于以上通信协议，实现了以下通信方式：

STDIO

采用STDIO的方式，server端会在client端启动时，作为client端的子进程一起启动。这种方式适用于client和server在同一台机器上通信的场景，通常用于工具调试。实现原理是client和server两个进程间通过stdin和stdout进行双向通信。

• 优点：无外部依赖；进程间通信极快；脱机可用
• 缺点：并发能力差，是同步阻塞模型；不支持多进程通信

SSE（Server-Sent Events）

一种基于 HTTP 协议的服务端主动推送机制，客户端通过 EventSource 向服务器发起一个长连接，服务器可以持续不断地通过这个连接发送文本数据。一般用于client在本地，server在远程服务器的场景。

• 优点：简单易用，浏览器原生支持 EventSource；自动重连，断线后自动尝试重新连接
• 缺点：单向通信（只能服务器推送）；不支持二进制，只能发送 UTF-8 文本；浏览器兼容性差；网络问题难以感知；不支持自定义 Header，不便于鉴权

Streamable HTTP

本质上仍是标准的 HTTP 请求，但服务器端采用分块传输（chunked transfer encoding），可以在响应体中逐块输出数据，实现流式返回。

SSE 虽然在早期实现服务端推送场景中提供了一种简洁方式，但随着 AI 模型接口对可靠性、跨平台性、鉴权机制以及链路可控性提出更高要求，SSE 的局限性逐渐暴露。而 Streamable HTTP 凭借其通用性、稳定性和良好的生态支持，成为更适合 AI 场景的流式传输方案。

• 优点：支持流式响应，兼容所有现代浏览器和 HTTP 客户端；与标准 HTTP 完全兼容；支持自定义 Header；可在客户端感知中断；不依赖浏览器特性
• 缺点：实现稍复杂；不像 WebSocket 那样全双工

工作流程

以下是MCP完成一次工作的完整示例：

1. 连接：建立通信通道 — MCP客户端启动并查找可用的MCP服务器。服务器验证请求，建立通信通道。连接建立后，客户端可以获取可用的资源、工具和提示信息。主机应用可以同时连接多个MCP服务器。
2. 发送请求：主机请求数据或操作 — 当用户在AI应用中提出请求，MCP客户端会解析用户输入，识别任务类型，然后选择合适的MCP服务器发送请求。
3. 处理请求：服务器执行操作 — 服务器收到请求后，会执行相应的操作，可能涉及到访问本地数据源、调用远程API、执行计算任务或者组合多个数据源提供综合信息。
4. 返回结果：服务器将响应发送回主机 — MCP服务器完成请求处理后，会将结果打包并发送回MCP客户端。
5. 生成响应：AI处理数据并反馈给用户 — MCP客户端收到服务器返回的数据后，会将其传递给AI应用进行进一步处理。
6. 断开连接（可选） — 任务完成后可断开，或保持连接以随时处理新请求。

---

5.2 MCP服务器

MCP 服务器介绍

MCP 服务端是什么

角色定位：MCP Server 是跑在本地或远程的"轻量级能力插件"，向上为 LLM/Agent 暴露标准化的 资源（Resources）、工具（Tools）、提示（Prompts）三类能力，向下则安全地访问文件、数据库、API 等实际数据源。

架构位置：整个 MCP 采用经典 C/S 架构——Host（Claude Desktop、IDE 等）→ MCP Client（1:1 连接）→ MCP Server → 本地/远程资源。

服务端种类

当下主流的与大模型交互的三要素无非是：工具、资源、提示词，而MCP针对这三类均做了标准化处理：

• Resources：定制化地请求和访问本地的资源，可以是文件系统、数据库、当前代码编辑器中的文件等等原本网页端的app无法访问到的静态资源。额外的 resources 会丰富发送给大模型的上下文，使得 AI 给我们更加精准的回答。
• Prompts：定制化一些场景下可供 AI 进行采纳的 prompt，比如如果需要 AI 定制化地返回某些格式化内容时，可以提供自定义的 prompts。
• Tools：可供 AI 使用的工具，它必须是一个函数，比如预定酒店、打开网页、关闭台灯这些封装好的函数就可以是一个 tool，大模型会通过 function calling 的方式来使用这些 tools。Tools 将会允许 AI 直接操作我们的电脑，甚至和现实世界发生交互。

UV 工具介绍

MCP开发要求借助uv进行虚拟环境创建和依赖管理。uv 是一个Python 依赖管理工具，类似于 pip 和 conda，但它更快、更高效，并且可以更好地管理 Python 虚拟环境和依赖项。它的核心目标是替代 pip、venv 和 pip-tools，提供更好的性能和更低的管理开销。

MCP 服务器开发

功能说明

以天气查询助手为例演示如何进行开发。

代码编写

import json
import os
from typing import Any
import httpx
import dotenv
from mcp.server.fastmcp import FastMCP
from loguru import logger

dotenv.load_dotenv()

mcp = FastMCP("WeatherServer")


@mcp.tool()
async def get_weather(city: str) -> dict[str, Any] | None:
    """
    查询指定城市的即时天气信息。

    :param city: 必要参数，字符串类型，表示要查询天气的城市名称。
                 注意：中国城市需使用其英文名称，如 "Beijing" 表示北京。
    :return: 返回 OpenWeather API 的响应结果。
    """
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": os.getenv("OPENWEATHER_API_KEY"),
        "units": "metric",
        "lang": "zh_cn"
    }

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, params=params, timeout=30.0)
            response.raise_for_status()
            logger.info(f"查询天气结果：{json.dumps(response.json())}")
            return response.json()
        except Exception as e:
            logger.error(f"查询天气失败：{e}")
            return None


if __name__ == "__main__":
    logger.info("启动 MCP 服务器...")
    mcp.run(transport='stdio')

运行服务

在本地启动一个 MCP 服务器，运行刚刚创建的 server.py 文件，并进入一个开发模式，方便：

• 热加载代码（修改后不用重启整个客户端）
• 在终端里查看服务器和 MCP 客户端之间的通信日志（方便调试）
• 模拟 MCP 客户端连接到你的服务器，测试 API / 工具调用是否正常

运行 Inspector：

# npx -y @modelcontextprotocol/inspector uv run server.py
Starting MCP inspector...
⚙️ Proxy server listening on localhost:6277
🔑 Session token: c74f080a0d770ea7ae26594304ad9cc160a1a131f54f7b9db3a89f98d25cb672
   Use this token to authenticate requests or set DANGEROUSLY_OMIT_AUTH=true to disable auth

🚀 MCP Inspector is up and running at:
   http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=c74f080a0d770ea7ae26594304ad9cc160a1a131f54f7b9db3a89f98d25cb672

🌐 Opening browser...

也可以执行：

mcp dev server.py

服务启动后会自动打开浏览器工具。

使用 MCP 服务器

为方便测试 MCP 服务器，暂时先不开发 MCP 客户端，而是通过第三方客户端工具调用。

安装 Cherry Studio

具体可参考文档：https://docs.cherry-ai.com/advanced-basic/mcp/config

配置 MCP 服务器

新增一个 MCP 服务器，启动参数如下：

--directory
D:\PycharmProjects\LangChainDemo  # 替换为实际项目路径
run
server.py

运行测试

在 MCP 工具中启用天气查询工具，然后进行访问测试。

原理分析

启用 MCP 后，Cherry Studio 不只是把用户的输入直接交给 Ollama，它会多一个 “模型中转+工具编排” 的流程：

1. 用户输入 — 你在 Cherry Studio 输入问题
2. Cherry Studio 请求模型（前置 MCP 拦截） — Cherry Studio 收到输入后，并不会立刻把它丢给 Ollama，而是先通过 MCP Client 传递给 MCP Server。MCP Server 里定义了可用的 Tools、Prompts、Resources。
3. 模型生成调用计划（Tool Call） — MCP Server 会先把你的输入传给 Ollama 模型（作为推理引擎）。模型可能返回一个工具调用意图（比如 call weather_tool(city="北京")）。
4. MCP 调用外部工具 — MCP Server 检查调用请求：工具调用 → 执行对应的 Tool；Resource 读取 → 拉取外部文件/数据；Prompt 模板 → 组合成最终请求。工具执行完成后，把结果（比如天气 JSON）返回给 MCP Server。
5. 再次调用模型生成最终回答 — MCP Server 把工具返回的结果再次传给 Ollama 模型，让它生成最终的自然语言回答。
6. Cherry Studio 显示对话结果 — Cherry Studio 渲染模型返回的最终结果到聊天窗口。

---

5.3 MCP客户端

MCP 客户端开发

客户端示例

初始化 MCP 客户端（但不连接服务器），并提供一个交互式 CLI，可以输入查询（但只返回模拟回复），通过输入 quit 退出程序。需要注意的是，此时客户端没有关联任何大模型，因此只会重复用户的输入。

from loguru import logger
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()

    async def connect_to_mock_server(self):
        logger.info("✅ MCP 客户端已初始化，但未连接到服务器")

    async def chat_loop(self):
        logger.info("MCP 客户端已启动！")
        print("输入你的问题或输入 'quit' 退出。")
        while True:
            try:
                query = input("\n🧑‍🦲 [用户输入]: ").strip()
                if query.lower() == 'quit':
                    break
                print(f"\n🤖 [AI回答] ：{query}")
            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")

    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        await client.connect_to_mock_server()
        await client.chat_loop()
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

接入在线 AI 模型

获取 API Key

以阿里云百炼为例，登录 https://bailian.console.aliyun.com/?tab=model#/api-key ，获取 API Key。

创建 .env 文件

OPEN_API_KEY="XXXXXXXXXXXXXX"
BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL="deepseek-r1-distill-llama-70b"

修改 client.py 代码

import os
from openai import OpenAI
from loguru import logger
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from dotenv import load_dotenv
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam

load_dotenv()


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.base_url = os.getenv("BASE_URL")
        self.openai_api_key = os.getenv("OPEN_API_KEY")
        self.model = os.getenv("MODEL")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)

    async def process_query(self, query: str) -> str:
        messages = [
            ChatCompletionSystemMessageParam(role="system", content="你是一个智能助手，帮助用户回答问题。"),
            ChatCompletionUserMessageParam(role="user", content=query)
        ]
        try:
            response = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=self.model,
                    messages=messages
                )
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"⚠️ 调用 OpenAI API 时出错: {str(e)}"

    async def chat_loop(self):
        logger.info("MCP 客户端已启动！")
        print("输入你的问题或输入 'quit' 退出。")
        while True:
            try:
                query = input("\n🧑‍🦲 [用户输入]: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)
                print(f"\n🤖 [AI回答] ：{response}")
            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")

    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        await client.chat_loop()
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

接入本地 AI 模型

由于ollama和vLLM均支持OpenAI API风格调用方法，因此上述 client.py 并不需要进行任何修改，只需要启动相应的调度框架服务，然后修改 .env 文件即可。

• ollama 部署参考：https://www.cuiliangblog.cn/detail/section/227776360
• vLLM 部署参考：https://www.cuiliangblog.cn/detail/section/227776424

修改 .env 文件：

OPEN_API_KEY=""
BASE_URL="http://127.0.0.1:11434/v1"
MODEL="qwen3:14b"

接入 MCP 服务器

修改客户端文件

import json
import os
import sys
from openai import OpenAI
from loguru import logger
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters, stdio_client
from dotenv import load_dotenv
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam

load_dotenv()


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.base_url = os.getenv("BASE_URL")
        self.openai_api_key = os.getenv("OPEN_API_KEY")
        self.model = os.getenv("MODEL")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)

    async def connect_to_server(self, server_script_path: str):
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("服务器脚本必须是 .py 或 .js 文件")

        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )

        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        await self.session.initialize()

        response = await self.session.list_tools()
        tools = response.tools
        logger.info(f"已连接到服务器，支持以下工具:{[tool.name for tool in tools]}")

    async def process_query(self, query: str) -> str:
        messages = [
            ChatCompletionSystemMessageParam(role="system", content="你是一个智能助手，帮助用户回答问题。"),
            ChatCompletionUserMessageParam(role="user", content=query)
        ]

        response = await self.session.list_tools()
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=available_tools
        )

        content = response.choices[0]
        if content.finish_reason == "tool_calls":
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            result = await self.session.call_tool(tool_name, tool_args)
            logger.info(f"[调用工具] {tool_name} 传入参数是: {tool_args}")

            messages.append(content.message.model_dump())
            messages.append({
                "role": "tool",
                "content": result.content[0].text,
                "tool_call_id": tool_call.id,
            })

            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
            )
            return response.choices[0].message.content

        return content.message.content

    async def chat_loop(self):
        logger.info("MCP 客户端已启动！")
        print("输入你的问题或输入 'quit' 退出。")
        while True:
            try:
                query = input("\n🧑‍🦲 [用户输入]: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)
                print(f"\n🤖 [AI回答] ：{response}")
            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")

    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        if len(sys.argv) < 2:
            logger.error("请提供 MCP server 脚本路径，例如：python client.py server.py")
            return
        await client.connect_to_server('server.py')
        await client.chat_loop()
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

运行演示

uv run client.py server.py

运行后，客户端会连接到 MCP 服务器，获取可用工具列表（如 get_weather），然后进入交互式对话。当用户提问"上海天气怎么样"时，模型会自动调用天气工具获取数据，并生成自然语言回答。

---

5.4 MCP进阶使用

基于 SSE 的 MCP 实现

除了stdio连接模式外，MCP还提供了可以服务器、客户端异地运行的SSE传输模式，以适用于更加通用的开发情况。

Server 端代码

import json
import os
import httpx
import dotenv
from mcp.server.fastmcp import FastMCP
from loguru import logger

dotenv.load_dotenv()

mcp = FastMCP("WeatherServerSSE", host="0.0.0.0", port=8000)

@mcp.tool()
def get_weather(city: str) -> dict | None:
    """
    查询指定城市的即时天气信息。
    参数 city: 城市英文名，如 Beijing
    """
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": os.getenv("OPENWEATHER_API_KEY"),
        "units": "metric",
        "lang": "zh_cn"
    }

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, params=params, timeout=30.0)
            response.raise_for_status()
            logger.info(f"查询天气结果：{json.dumps(response.json())}")
            return response.json()
        except Exception as e:
            logger.error(f"查询天气失败：{e}")
            return None

if __name__ == "__main__":
    logger.info("启动 MCP SSE 天气服务器，监听 http://0.0.0.0:8000/sse")
    mcp.run(transport="sse")

Server 端测试

使用 Cherry Studio 来调用这个 MCP Server。打开 Cherry Studio 的 MCP 添加配置界面，类型选择 SSE，url 填写 http://localhost:8000/sse。SSE 类型的 MCP Server 配置起来要比 stdio 类型要简单很多，只需要配个 url 即可。

之后回到聊天界面，依然问一个需要查询天气的问题

Client 端代码

import json
import os
import sys
from mcp.client.sse import sse_client
from openai import OpenAI
from loguru import logger
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from dotenv import load_dotenv
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam

load_dotenv()

class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.base_url = os.getenv("BASE_URL")
        self.openai_api_key = os.getenv("OPEN_API_KEY")
        self.model = os.getenv("MODEL")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)

    async def connect_to_server(self, sse_url):
        sse_transport = await self.exit_stack.enter_async_context(sse_client(sse_url))
        self.session = await self.exit_stack.enter_async_context(ClientSession(*sse_transport))
        await self.session.initialize()
        tools = (await self.session.list_tools()).tools
        logger.info(f"已连接 SSE 服务器，支持工具: {[t.name for t in tools]}")

    async def process_query(self, query: str) -> str:
        messages = [
            ChatCompletionSystemMessageParam(role="system", content="你是一个智能助手，帮助用户回答问题。"),
            ChatCompletionUserMessageParam(role="user", content=query)
        ]

        response = await self.session.list_tools()
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=available_tools
        )

        content = response.choices[0]
        if content.finish_reason == "tool_calls":
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            result = await self.session.call_tool(tool_name, tool_args)
            logger.info(f"[调用工具] {tool_name} 传入参数是: {tool_args}")

            messages.append(content.message.model_dump())
            messages.append({
                "role": "tool",
                "content": result.content[0].text,
                "tool_call_id": tool_call.id,
            })

            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
            )
            return response.choices[0].message.content

        return content.message.content

    async def chat_loop(self):
        logger.info("MCP 客户端已启动！")
        print("输入你的问题或输入 'quit' 退出。")
        while True:
            try:
                query = input("\n🧑‍🦲 [用户输入]: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)
                print(f"\n🤖 [AI回答] ：{response}")
            except Exception as e:
                print(f"\n⚠️ 发生错误: {str(e)}")

    async def cleanup(self):
        await self.exit_stack.aclose()

async def main():
    client = MCPClient()
    sse_url = "http://localhost:8000/sse"
    try:
        await client.connect_to_server(sse_url)
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

client验证

输入你的问题或输入 'quit' 退出。

🧑‍🦲 [用户输入]: 广州天气怎么样
2025-08-14 15:20:05.323 | INFO     | __main__:process_query:102 - [调用工具] get_weather 传入参数是: {'city': 'Guangzhou'}

🤖 [AI回答] ：<think>
好的，用户之前询问了广州的天气，现在我需要根据提供的天气数据来生成回答。首先，查看数据中的主要信息，比如温度、天气状况、风力等。温度方面，当前温度是25.91°C，体感温度26.93°C，湿度91%，降雨量1小时内4.09毫米，说明有大雨。风速是3.48m/s，方向164度，可能来自东南方向。云量是100%，表示完全被云覆盖，天气阴沉。此外，日出和日落时间可能对用户有帮助，特别是如果他们计划外出活动的话。需要将这些信息组织成自然的中文回答，确保用户能清楚了解当前的天气情况，并给出适当的建议，比如携带雨具或注意防雨。同时，保持回答简洁明了，避免使用过多技术术语，让用户容易理解。
</think>

广州当前天气为大雨，气温25.91°C，体感温度26.93°C，湿度高达91%。1小时内降雨量达4.09毫米，建议外出携带雨具。风力3.48m/s，东南风方向，天空被云层完全覆盖，日出时间19:52，日落时间18:41，昼夜温差较小，注意防雨防滑。

基于 Streamable HTTP 的 MCP 实现

相比SSE传输，HTTP流式传输并发更高、通信更加稳定，同时也更容易集成和部署，这也是当代服务器与客户端异地通信的最佳解决方案。在5月9号的1.8.0版本更新中，正式在SDK中加入了HTTP流式MCP服务器的相关功能支持。

以天气查询服务器为例：

import os, json, contextlib
import click, httpx, dotenv, uvicorn
from loguru import logger
from collections.abc import AsyncIterator
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Receive, Scope, Send
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager

dotenv.load_dotenv()

async def fetch_weather(city: str) -> dict | None:
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": os.getenv("OPENWEATHER_API_KEY"),
        "units": "metric",
        "lang": "zh_cn",
    }
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            res = await client.get(url, params=params)
            res.raise_for_status()
            logger.info(f"获取天气数据结果: {res.json()}")
            return res.json()
    except Exception as e:
        logger.error(f"天气查询失败: {e}")
        return None

@click.command()
@click.option("--port", default=3000, help="Port to listen on for HTTP")
def main(port: int):
    app = Server("mcp-weather")

    @app.call_tool()
    async def get_weather(name: str, arguments: dict) -> list[types.TextContent]:
        city = arguments.get("location")
        if not city:
            raise ValueError("'location' is required")

        ctx = app.request_context
        await ctx.session.send_log_message("info", f"Fetching weather for {city}…",
                                           logger="weather", related_request_id=ctx.request_id)
        weather = await fetch_weather(city)
        if not weather:
            raise RuntimeError("获取天气数据失败")

        await ctx.session.send_log_message("info", "Weather data fetched successfully!",
                                           logger="weather", related_request_id=ctx.request_id)
        return [types.TextContent(type="text", text=json.dumps(weather, ensure_ascii=False, indent=2))]

    @app.list_tools()
    async def list_tools() -> list[types.Tool]:
        return [types.Tool(
            name="get-weather",
            description="查询指定城市的实时天气（OpenWeather 数据）",
            inputSchema={
                "type": "object",
                "required": ["location"],
                "properties": {
                    "location": {"type": "string", "description": "城市的英文名称，如 'Beijing'"},
                },
            },
        )]

    session_manager = StreamableHTTPSessionManager(app=app, event_store=None, stateless=True)

    async def handle(scope: Scope, receive: Receive, send: Send) -> None:
        await session_manager.handle_request(scope, receive, send)

    @contextlib.asynccontextmanager
    async def lifespan(_: Starlette) -> AsyncIterator[None]:
        async with session_manager.run():
            logger.info("Weather MCP server started")
            yield
            logger.info("Weather MCP server shutting down…")

    starlette_app = Starlette(debug=False, routes=[Mount("/mcp", app=handle)], lifespan=lifespan)
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)

if __name__ == "__main__":
    main()

server 端测试

使用 Cherry Studio 调用时，类型选择流式传输 HTTP，url 填写 http://localhost:3000/mcp/。

之后回到聊天界面，依然问一个需要查询天气的问题

Resources、Prompt 类功能 MCP 服务器

除了Tools功能的服务器外，MCP还支持Resources类服务器和Prompt类服务器：

• Resources服务器：主要负责提供更多的资源接口，如调用本地文件、数据等
• Prompt类服务器：用于设置Agent开发过程中各环节的提示词模板

公开 & 在线 MCP 调用

MCP标准通信协议带来的最大价值之一，就是让广大Agent开发者能够基于此进行协作。几个有名的MCP服务器合集（导航站）地址：

• MCP官方服务器合集：https://github.com/modelcontextprotocol/servers
• MCP Github热门导航：https://github.com/punkpeye/awesome-mcp-servers
• MCP工具注册平台：https://github.com/ahujasid/blender-mcp
• MCP导航：https://mcp.so/
• 魔搭 MCP：https://www.modelscope.cn/mcp
• 阿里云百炼：https://bailian.console.aliyun.com/?tab=mcp

Client 进阶功能

除了能在命令行中创建MCP客户端外，还支持各类客户端的调用：https://modelcontextprotocol.io/clients

借助对话类客户端（如 Claude Desktop），可以轻易地将各类服务器进行集成。在 IDE 客户端（如 Cline 或 Cursor）中，可以直接调用服务器进行开发。

此外，还有一些为MCP量身定制的Agent开发框架：https://github.com/lastmile-ai/mcp-agent

---

5.5 Cursor接入MCP

Cursor 安装与使用

具体可参考文档：https://www.cuiliangblog.cn/detail/section/232745699

Cursor接入MCP的方法有很多种，可以将Smithery平台上的MCP工具或GitHub MCP工具接入Cursor。

访问地址：https://smithery.ai/

Smithery 与 GitHub 对比

Smithery 是一个专门用于管理和分发 MCP 服务器的平台。

托管方式：

• Smithery 平台：提供远程（Remote）/ 托管（Hosted）和 本地（Local）两种模式
• GitHub：主要提供 MCP 服务器的源代码，开发者需要自行下载、配置并运行

安装与管理：

• Smithery 平台：提供统一的界面和 CLI 工具，简化了发现、安装和管理过程
• GitHub：需要手动克隆仓库、安装依赖项，过程相对繁琐

安全性与控制：

• Smithery 平台：配置参数（如访问令牌）是临时的，不会长期存储在其服务器上
• GitHub：开发者完全控制代码和运行环境，可以自行审查代码

社区与支持：

• Smithery 平台：作为 MCP 服务器的集中管理平台，聚集了大量 MCP 服务器
• GitHub：拥有广泛的社区支持

安装基础依赖

在使用 MCP 时，是否需要安装 Node.js 取决于所选择的 MCP 服务器的实现方式。目前许多开发者选择使用 Node.js 来实现 MCP 服务器，因为其拥有丰富的包管理生态系统（npm）以及在处理异步操作和 I/O 密集型任务方面的高效性。

安装完成后检查：

node -v
npx -v
uv -V

添加 MCP 常用工具

Sequential Thinking

Sequential Thinking 是一个基于 MCP 的服务器工具，旨在通过结构化的思维流程，帮助用户动态、反思性地解决复杂问题。该工具将问题拆解为可管理的步骤，每个步骤都可以建立在先前的见解之上，或对其进行质疑和修正。

插件地址：https://modelscope.cn/mcp/servers/@modelcontextprotocol/sequentialthinking

然后打开cursor，添加 MCP：

按照提示将配置粘贴进文件

安装完成后，等待验证：

然后进行简单问答测试，查看能否顺利调用工具：

Playwright

Playwright Automation 是一个基于 MCP 的服务器工具，利用 Playwright 为 LLMs 和 AI 助手提供与网页交互的能力。

主要功能：

• 网页导航与交互：自动执行网页导航、点击、表单填写等操作
• 内容提取与网页抓取：从网页中提取结构化数据
• 截图与可视化：捕获网页或特定元素的截图
• JavaScript 执行：在浏览器环境中执行自定义 JavaScript 代码

工具主页：https://modelscope.cn/mcp/servers/@microsoft/playwright-mcp

然后打开cursor，添加 MCP：将 json 内容粘贴至 mcp.json 文件中

访问验证：

FileSystem

Filesystem MCP 是一个为 LLMs 和 AI 助手提供对本地文件系统的安全、受控访问的服务器工具。

主要功能：

• 文件读写：允许读取和写入文件内容
• 目录管理：支持创建、列出和删除目录
• 文件搜索：能够搜索匹配特定模式的文件或目录
• 元数据获取：获取文件或目录的详细元数据

工具地址：https://modelscope.cn/mcp/servers/@modelcontextprotocol/filesystem

配置示例：

"filesystem": {
  "command": "npx",
  "args": [
    "-y",
    "@modelcontextprotocol/server-filesystem",
    "C://Users//桌面",
    "D://其他路径"
  ]
}

然后进行测试

Fetch

Fetch MCP Server 是一种专门为语言模型设计的 MCP 服务器，用于抓取网页内容并将 HTML 转换成 AI 模型更易处理的 Markdown 格式。

工具地址：https://modelscope.cn/mcp/servers/@modelcontextprotocol/fetch

获取安装命令

配置示例：

"fetch": {
  "command": "uvx",
  "args": ["mcp-server-fetch"]
}

测试验证