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

# 创建项目目录 mkdir my-mcp-server cd my-mcp-server # 创建虚拟环境 python -m venv venv source venv/bin/activate # 在Windows上：venv\Scripts\activate # 安装依赖 pip install mcp httpx 

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)

@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 

@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}} 提供具体的改进建议，并附上代码示例。 """ ) ] 

async def search_internal_docs(query: str) -> list[types.TextContent]: # 实现搜索内部文档的功能 # 这可以连接到公司Wiki、Confluence或其他系统 pass 

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 

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="访问被拒绝：无效目录")]