构建您自己的MCP服务器：连接Cursor的终极指南

大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

模型上下文协议（MCP）是AI领域的USB-C——一种正在改变我们与大型语言模型（LLM）交互方式的通用连接器。2024年底由Anthropic发布，MCP允许开发人员构建AI模型与外部工具或数据源之间的安全双向连接。

MCP服务器使您能够在保持对数据完全控制的同时，为Claude、Cursor、Windsurf等编辑器扩展个性化功能。通过连接 MCP 服务器，显著提升了运行的稳定性和流畅性，简化了开发流程，通过集成多种外部工具和服务（如数据库管理、UI/UX设计、浏览器自动化等）扩展了功能，并增强了代码生成的准确性和相关性。此外，MCP协议支持多语言和跨平台，确保了数据的安全性和合规性，从而提高了整体工作效率和开发体验。

通过MCP，使得AI编程工具，例如Cursor、Windsurf、Claude Code和Zed，现在可以访问：

您的本地文件和数据库
您创建的自定义工具
第三方API和服务
私有文档

同时将敏感数据保留在您的基础设施内。

让我们构建一个简单但实用的MCP服务器，任何兼容的客户端都可以连接到它。我们将创建一个具有以下功能的服务器：

智能搜索您的代码库
执行自定义代码分析
连接到外部开发API

Python 3.10+或Node.js 16+
基本了解JSON-RPC
一个MCP兼容的客户端（如Claude Desktop、Cursor、Windsurf、Zed等）

我们将使用Python和官方MCP SDK来完成这个示例。创建一个新的项目目录并设置您的环境：

# 创建项目目录 mkdir my-mcp-server cd my-mcp-server

创建虚拟环境

python -m venv venv source venv/bin/activate # 在Windows上：venv\Scripts\activate

安装依赖

pip install mcp httpx

创建一个名为server.py的文件，并添加以下代码：

from mcp.server.lowlevel import Server import mcp.types as types import httpx import os import json import glob

初始化服务器

app = Server(“dev-tools-server”)

@app.call_tool() asyncdef call_tool(name: str, arguments: dict) -> list[types.TextContent]:

"""处理来自客户端的工具调用。""" if name == "search_code": returnawait search_code(arguments.get("query", ""), arguments.get("directory", "")) elif name == "analyze_dependencies": returnawait analyze_dependencies(arguments.get("directory", ".")) elif name == "fetch_documentation": returnawait fetch_documentation(arguments.get("package", "")) else: return [types.TextContent(type="text", text=f"错误：未知工具：{name}")]

asyncdef search_code(query: str, directory: str) -> list[types.TextContent]:

"""在代码文件中搜索特定查询内容。"""
results = []
for ext in ['.py', '.js', '.ts', '.jsx', '.tsx', '.html', '.css']:
    for filepath in glob.glob(f"{directory}//*{ext}", recursive=True):
        try:
            with open(filepath, 'r', encoding='utf-8') as file:
                content = file.read()
            if query.lower() in content.lower():
                match_context = get_context(content, query)
                results.append(f"文件：{filepath}\n{match_context}\n--")
        except Exception as e:
            continue
if results:
    return [types.TextContent(type="text", text="搜索结果：\n\n" + "\n".join(results))]
else:
    return [types.TextContent(type="text", text=f"在目录'{directory}'中未找到与'{query}'匹配的结果。")]

def get_context(content: str, query: str, context_lines: int = 3) -> str:

"""获取匹配项周围的上下文。""" lines = content.split('\n') matches = [] for i, line in enumerate(lines): if query.lower() in line.lower(): start = max(0, i - context_lines) end = min(len(lines), i + context_lines + 1) context = "\n".join(lines[start:end]) matches.append(f"行{start+1}-{end}:\n{context}") return"\n\n".join(matches)

asyncdef analyze_dependencies(directory: str) -> list[types.TextContent]:

"""分析项目依赖项。""" dependency_files = { 'python': ['requirements.txt', 'setup.py', 'pyproject.toml'], 'node': ['package.json'], 'dotnet': ['*.csproj', '*.fsproj', '*.vbproj'], } results = [] for lang, files in dependency_files.items(): for file_pattern in files: for filepath in glob.glob(f"{directory}//{file_pattern}", recursive=True): try: with open(filepath, 'r', encoding='utf-8') as file: content = file.read() results.append(f"在{filepath}中发现{lang}依赖项") except Exception: continue if results: return [types.TextContent(type="text", text="依赖项分析：\n\n" + "\n".join(results))] else: return [types.TextContent(type="text", text=f"在目录'{directory}'中未找到依赖项文件。")]

asyncdef fetch_documentation(package: str) -> list[types.TextContent]:

"""获取包的文档。""" try: # 对于Python包 url = f"https://pypi.org/pypi/{package}/json" asyncwith httpx.AsyncClient() as client: response = await client.get(url, timeout=10.0) if response.status_code == 200: data = response.json() summary = data.get("info", {}).get("summary", "无摘要可用") description = data.get("info", {}).get("description", "无描述可用") return [types.TextContent(type="text", text=f"{package}的文档：\n\n摘要：{summary}\n描述：{description}")] except Exception: pass # 如果PyPI失败，尝试npm包 try: url = f"https://registry.npmjs.org/{package}" asyncwith httpx.AsyncClient() as client: response = await client.get(url, timeout=10.0) if response.status_code == 200: data = response.json() description = data.get("description", "无描述可用") return [types.TextContent(type="text", text=f"{package}的文档：\n\n描述：{description}")] except Exception: pass return [types.TextContent(type="text", text=f"无法获取'{package}'的文档。")]

@app.list_tools() asyncdef list_tools() -> list[types.Tool]:

"""列出可用工具。""" return [ types.Tool( name="search_code", description="在代码文件中搜索特定查询内容", inputSchema={ "type": "object", "required": ["query"], "properties": { "query": {"type": "string", "description": "要在代码文件中搜索的查询内容"}, "directory": {"type": "string", "description": "搜索目录（默认：当前目录）"} } } ), types.Tool( name="analyze_dependencies", description="分析项目依赖项", inputSchema={ "type": "object", "properties": { "directory": {"type": "string", "description": "分析目录（默认：当前目录）"} } } ), types.Tool( name="fetch_documentation", description="获取包的文档", inputSchema={ "type": "object", "required": ["package"], "properties": { "package": {"type": "string", "description": "要获取文档的包名"} } } ) ]

if name == “main”:

import sys # 默认使用标准输入输出传输 transport = "stdio" port = 8000 # 检查命令行参数 if len(sys.argv) > 1: if sys.argv[1] == "sse": transport = "sse" if len(sys.argv) > 2: try: port = int(sys.argv[2]) except ValueError: pass if transport == "sse": from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Mount, Route import uvicorn sse = SseServerTransport("/messages/") asyncdef handle_sse(request): asyncwith 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), ] ) print(f"在端口{port}上启动MCP服务器，使用SSE传输") uvicorn.run(starlette_app, host="0.0.0.0", port=port) else: from mcp.server.stdio import stdio_server import anyio asyncdef run_stdio(): asyncwith stdio_server() as streams: await app.run(streams[0], streams[1], app.create_initialization_options()) print("使用标准输入输出传输启动MCP服务器") anyio.run(run_stdio)

您可以以两种模式运行服务器：

标准输入输出模式（适用于桌面客户端，如Claude Desktop）：python server.py
服务器发送事件（SSE）模式（适用于基于Web的客户端）：python server.py sse 8000

现在您已经有一个可以工作的MCP服务器，让我们将它连接到不同的客户端：

编辑您的Claude Desktop配置文件：

在macOS上：~/Library/Application Support/Claude/claude_desktop_config.json
在Windows上：%APPDATA%\Claude\claude_desktopconfig.json
”>添加您的服务器配置：{
“mcpServers”: {
“dev-tools”: {
“command”: “python”,
“args”: [“/absolute/path/to/my-mcp-server/server.py”]
}
}
}
重启Claude Desktop并查找工具图标出现。

对于Cursor，您通常会使用SSE传输：

以SSE模式运行您的服务器：python server.py sse 8000
在Cursor中，转到设置 → 功能 → MCP服务器
添加一个新的服务器，配置如下：

名称：dev-tools
类型：sse
URL：http://localhost:8000/sse

大多数较新的客户端遵循与Cursor类似的配置模式：

以SSE模式运行您的服务器
在客户端的设置中查找MCP配置
使用SSE URL进行配置

让我们为服务器添加一些更强大的功能：

MCP服务器可以向客户端提供文件资源。这允许AI模型直接访问和读取文件：

@app.list_resources() asyncdef list_resources() -> list[types.Resource]:

"""列出可用资源。""" resources = [] # 添加README文件（如果存在） readme_paths = ['README.md', 'README.txt', 'Readme.md'] for path in readme_paths: if os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: content = f.read() resources.append(types.Resource( name="readme", description="项目README文件", content=[types.TextContent(type="text", text=content)] )) break return resources

MCP还支持提示模板，这些是引导模型响应的预写模板：

@app.list_prompts() asyncdef list_prompts() -> list[types.Prompt]:

"""列出可用提示。""" return [ types.Prompt( name="code_review", description="对特定文件进行代码审查", inputSchema={ "type": "object", "required": ["filepath"], "properties": { "filepath": { "type": "string", "description": "要审查的文件路径" }, "focus": { "type": "string", "description": "要重点关注的具体方面（例如，'安全性'）" } } }, template=""" 请审查以下来自{{filepath}}的代码： {{#file filepath}} {{#if focus}} 专注于{{focus}}方面的审查。 {{else}} 进行一般性代码审查，重点关注： - 代码质量 - 潜在错误 - 性能问题 - 安全问题 {{/if}} 提供具体的改进建议，并附上代码示例。 """ ) ]

在开发MCP服务器时，请牢记以下安全实践：

输入验证：始终验证并清理来自客户端的输入。
文件访问控制：限制文件访问到特定目录。
API密钥处理：不要将API密钥硬编码；使用环境变量。
速率限制：为资源密集型操作实现速率限制。
错误处理：返回信息丰富但不泄露过多细节的错误消息。

MCP服务器不仅仅是一个酷炫的技术演示——它可以彻底改变您的开发工作流程：

将您的MCP服务器连接到内部文档、代码库和知识库，创建一个真正了解您项目的个性化助手：

async def search_internal_docs(query: str) -> list[types.TextContent]:

# 实现搜索内部文档的功能 # 这可以连接到公司Wiki、Confluence或其他系统 pass

直接从AI助手自动化常见的开发任务：

async def run_tests(test_path: str = None) -> list[types.TextContent]:

"""运行项目测试，可选路径过滤。""" command = ["pytest"] if test_path: command.append(test_path) # 运行测试并捕获输出 # 将结果返回给AI pass

创建工具以在整个团队中强制执行代码质量标准：

async def lint_code(filepath: str) -> list[types.TextContent]:

"""对指定代码文件运行代码检查工具。""" # 实现运行代码检查工具并返回结果 pass

截至2025年3月，Claude Desktop和Cursor提供了强大的MCP支持，而Windsurf和Zed的实现可能在功能和稳定性方面有所不同。始终检查每个客户端的最新文档以获取当前兼容性信息。

提供的代码示例仅供演示目的，在生产环境中可能需要额外的错误处理和优化。请将其视为一个起点，而不是一个生产就绪的解决方案。

为了增强安全性，实现路径清理以防止目录遍历攻击：

import os.path directory = os.path.normpath(os.path.join(base_directory, user_input)) if not directory.startswith(base_directory):

return [types.TextContent(type="text", text="访问被拒绝：无效目录")]

对于Windsurf，MCP配置位于设置 → AI → 外部工具。

在将MCP服务器投入生产之前，务必使用您集成的每个客户端进行彻底测试。

MCP服务器代表了开发人员与AI交互方式的范式转变。通过构建您自己的MCP服务器，您创建了一个桥梁，将您的开发环境与AI助手连接起来，解锁了生产力和创造力的新可能性。

该技术仍在迅速发展，但早期采用者已经在工作流程中看到了显著的改进。无论您使用的是Claude、Cursor、Windsurf、Zed还是任何其他MCP兼容客户端，自定义MCP服务器都使您能够控制AI与您的开发环境的交互方式。

更多资源见星球，加入QuantML星球，与750+专业人士一起交流学习：

往期回顾

QuantML-Qlib开发版：

QuantML-Qlib重磅更新：DeepSeek核心模型结构用于选股
QuantML-Qlib Factor | 融合TA-Lib100+技术指标，自定义构建AlphaZoo
QuantML-Qlib Model | 还在使用MSE？试试这些更加适合金融预测的损失函数
QuantML-Qlib Model | 如何运行日内中高频模型
QuantML-Qlib Model | 超越GRU，液态神经网络LNN用于股票预测
QuantML-Qlib Model | 华泰SAM：提升AI量化模型的泛化性能研报复现
QuantML-Qlib Model | 华泰AlphaNet模型复现
QuantML-Qlib Model | 清华大学&华泰证券在高胜率时交易
QuantML-Qlib Factor | 高效优雅的因子构建方法：以开源金工切割动量因子为例
QuantML-Qlib Model | 滚动模型训练
QuantML-QlibModel | KAN + GRU 时序模型用于股票预测
QuantML-Qlib开发版 | 蚂蚁&清华 TimeMixer：可分解多尺度融合的时间序列模型用于金融市场预测
QuantML-Qlib Model | Kansformer：KAN+Transformer时序模型用于股票收益率预测
QuantML-QlibModel | 使用OPTUNA优化模型超参
QuantML-QlibDB | Clickhouse 行情存储与读取方案
QuantML-Qlib LLM | GPT-4o复现因子计算代码
QuantML-Qlib开发版 | 最新xLSTM用于股票市场预测
QuantML-Qlib开发版 | 强化学习因子挖掘
QuantML-Qlib开发版 | 清华大学时序SOTA模型iTransformer用于股票市场预测QuantML-Qlib开发版 | 最新神经网络结构KAN用于因子挖掘
QuantML-Qlib开发版 | 直接读取pg/mysql/mongodb数据库
QuantML-Qlib开发版 | MoE混合专家系统用于提升Transformer表现
QuantML-Qlib开发版 | 一键数据更新
QuantML-Qlib开发版 | AAAI**论文Informer用于金融市场预测
QuantML-Qlib开发版 | 取代Transformer的下一代神经网络结构Mamba用于金融市场预测
QuantML-Qlib开发版 | 时序SOTA模型PatchTST用于金融市场预测
QuantML-Qlib开发版 | 一行代码运行DLinear模型用于股票预测
研报复现：
重磅更新！80+量化策略复现（持续更新中）
BARRA CNE6模型复现
研报复现 | QRS择时信号及改进
研报复现 | 跳跃因子系列-下
研报复现 | 跳跃因子系列-上
研报复现 | 锚定反转因子
研报复现 | 另类ETF交易策略：日内动量
研报复现 | 国盛金工：如何将隔夜涨跌变为有效的选股因子？——基于对知情交易者信息优势的刻画
研报复现 | 招商证券：基于鳄鱼线的指数择时及轮动策略
研报复现 | 华西金工-股票网络与网络中心度因子研究
研报复现 | 基于筹码分布的选股策略
研报复现 | 开源金工-高频追涨杀跌因子复现
研报复现 | 开源证券：形态识别，均线的
券商研报因子复现及表现研究
前沿论文代码：
DeepSeek-TS+: MLA-Mamba及GRPO用于多序列预测统一框架
Hummingbot：开源加密货币做市机器人框架
FinRLlama：基于强化学习和市场反馈的金融情感分析LLM优化方案
端到端基于LLM的增强型交易系统
基于分层强化学习的日内风险因子挖掘
DeepScalper：深度强化学习捕捉日内交易的短暂机会
TradingAgents：基于多智能体LLM的金融交易框架
Kaggle - Optiver trading at the close第一名解决方案及部分代码
量化交易全攻略：从入门到精通的终极指南
普林斯顿&牛津大学 | 大模型在金融领域的应用、前景和挑战
Style Miner：基于强化学习算法的风格因子构造
AQR创始人Cliff Asness：市场效率下降假说
增强动量策略：动量Transformer模型
XGBoost 2.0 ：提升时间序列预测能力
NIPS 24 | FinCon: 基于LLM的多智能体交易及组合管理框架
NIPS 24 | CausalStock : 基于端到端因果发现的新闻驱动股价预测模型
JFE | 高效估计买卖价差的模型、实证与应用
超越传统网格交易：新型网格交易系统
JFE | ETF日内套利研究
NIPS 24 | 超越CVXPY,新型端到端优化器
揭秘Jane Street低延迟系统的优化技巧——减少系统抖动
南京大学LAMDA-强化学习DRL挖掘逻辑公式型Alpha因子
3万个因子，数据挖掘能超越同行审议的因子吗？
KDD 24 | 基于增强记忆的上下文感知强化学习的高频交易框架
FinRobot：用于金融领域的大模型AI平台
KDD 23 | DoubleAdapt: 显著提升各类模型表现的元学习模型
市场微观结构教程：深度订单簿预测
基于高频和日频因子的端到端直接排序组合构建模型
BOA 312页报告：Everything you wanted to know about quant
深度学习模型DeepLOB用于订单簿价格预测
What KAN I say？KAN代码全解析
取代MLP？MIT全新神经网络结构KAN,3天1.4k star
WWW‘24 | FinReport: 结合新闻语义信息的多因子模型显著提升预测准确性
WWW’24 | UniTime: 融合文本信息的时间序列预测模型
WWW‘24 | EarnMore: 如何利用强化学习来处理可定制股票池中的投资组合管理问题
KDD’23 | AlphaMix: 高效专家混合框架（MoE）显著提高上证50选股表现
IJCAI‘23 | StockFormer: RL+Self-Attention优化摆动交易提高股票预测精度
AAAI-24 | EarnHFT:针对高频交易的分层强化学习（RL）框架
AAAI-24 | MASTER 结合市场信息的自动特征选择的股票预测模型，25%年化收益
COLING 2024 | AlphaFin: 结合深度学习及大模型用于股票预测和金融问答，击败现有预测模型

构建您自己的MCP服务器：连接Cursor的终极指南

创建虚拟环境

安装依赖

初始化服务器

相关推荐