# 安装 MCP Python SDK（Anthropic 官方库）
pip install mcp

# 或者用 uv（推荐，比 pip 更快）
uv add mcp

python -c "import mcp; print(mcp.__version__)"

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import json
import re
from datetime import datetime

# 创建 Server 实例，名称会显示在 Claude 桌面版的工具列表里
app = Server("my-first-mcp-server")

# ===== 第一部分：注册工具列表 =====
# 这个函数告诉 Client："我有哪些工具，每个工具的名称、描述和参数是什么"

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="calculate",
            description="安全地执行数学计算，返回计算结果。"
                        "支持加减乘除和括号。适用于用户需要精确数学计算的场景。",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "要计算的数学表达式，只能包含数字和运算符，如 '2 + 3 * 4' 或 '(100 - 20) / 8'"
                    }
                },
                "required": ["expression"]
            }
        ),
        types.Tool(
            name="get_current_time",
            description="获取当前日期和时间。当用户询问现在几点、今天是几号时使用。",
            inputSchema={
                "type": "object",
                "properties": {}  # 无需任何参数
            }
        )
    ]

# ===== 第二部分：处理工具调用 =====
# 当 Client 调用工具时，这个函数被调用
# name: 工具名称（和 list_tools 里的 name 对应）
# arguments: 调用参数（字典格式）

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

    if name == "calculate":
        expression = arguments.get("expression", "")
        # 安全检查：只允许数字和基本运算符
        # 生产环境建议用 ast.parse 做更严格的检查
        if not re.match(r'^[\d\s\+\-\*\/\(\)\.]+$', expression):
            return [types.TextContent(
                type="text",
                text=f"错误：表达式包含不允许的字符。只支持数字、+、-、*、/、括号。"
            )]
        try:
            result = eval(expression)
            return [types.TextContent(
                type="text",
                text=f"{expression} = {result}"
            )]
        except ZeroDivisionError:
            return [types.TextContent(
                type="text",
                text="计算错误：除数不能为零"
            )]
        except Exception as e:
            return [types.TextContent(
                type="text",
                text=f"计算错误：{str(e)}"
            )]

    elif name == "get_current_time":
        now = datetime.now()
        return [types.TextContent(
            type="text",
            text=f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}（{now.strftime('%A')}）"
        )]

    else:
        return [types.TextContent(
            type="text",
            text=f"未知工具：{name}"
        )]

# ===== 第三部分：启动 Server =====
# 这是固定模板，不需要修改
# stdio_server 建立标准输入输出通信通道
# app.run 开始监听来自 Client 的请求

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

{
  "mcpServers": {
    "my-first-server": {
      "command": "python",
      "args": ["/Users/yourname/mcp_servers/server.py"]
    }
  }
}

用户：帮我计算 (123 + 456) * 7 等于多少

Claude 正在使用工具 calculate...
  输入：{"expression": "(123 + 456) * 7"}
  输出：(123 + 456) * 7 = 4053

Claude：(123 + 456) × 7 = 4053。

# file_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import os

app = Server("file-operations-server")

# 限制只能访问这个目录（安全考虑，防止 AI 访问系统文件）
ALLOWED_ROOT = os.path.expanduser("~/Documents/ai-workspace")
os.makedirs(ALLOWED_ROOT, exist_ok=True)

def safe_path(filename: str) -> str:
    """
    确保路径在允许的目录内
    防止路径遍历攻击：用户（或 AI）不能通过 '../../etc/passwd' 这类方式访问系统文件
    """
    # 计算规范化的绝对路径
    full_path = os.path.normpath(os.path.join(ALLOWED_ROOT, filename))
    # 检查是否在允许目录内
    if not full_path.startswith(ALLOWED_ROOT):
        raise ValueError(f"路径 '{filename}' 超出允许范围")
    return full_path

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="read_file",
            description="读取文件内容。只能访问 ~/Documents/ai-workspace 目录下的文件。",
            inputSchema={
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "文件名（相对路径，如 'report.txt' 或 'data/notes.md'）"
                    }
                },
                "required": ["filename"]
            }
        ),
        types.Tool(
            name="write_file",
            description="写入文件内容（会覆盖已有内容）。只能写入 ~/Documents/ai-workspace 目录。",
            inputSchema={
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "文件名（相对路径）"
                    },
                    "content": {
                        "type": "string",
                        "description": "要写入的文件内容"
                    }
                },
                "required": ["filename", "content"]
            }
        ),
        types.Tool(
            name="list_files",
            description="列出工作目录中的所有文件。用于查看有哪些文件可以操作。",
            inputSchema={
                "type": "object",
                "properties": {}
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == "read_file":
            path = safe_path(arguments["filename"])
            with open(path, 'r', encoding='utf-8') as f:
                content = f.read()
            char_count = len(content)
            return [types.TextContent(
                type="text",
                text=f"文件 '{arguments['filename']}' 内容（共 {char_count} 字符）：\n\n{content}"
            )]

        elif name == "write_file":
            path = safe_path(arguments["filename"])
            # 如果目录不存在，自动创建
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'w', encoding='utf-8') as f:
                f.write(arguments["content"])
            return [types.TextContent(
                type="text",
                text=f"文件 '{arguments['filename']}' 写入成功（{len(arguments['content'])} 字符）"
            )]

        elif name == "list_files":
            files = []
            for root, dirs, filenames in os.walk(ALLOWED_ROOT):
                # 跳过隐藏目录
                dirs[:] = [d for d in dirs if not d.startswith('.')]
                for filename in filenames:
                    if not filename.startswith('.'):
                        rel_path = os.path.relpath(
                            os.path.join(root, filename),
                            ALLOWED_ROOT
                        )
                        files.append(rel_path)

            if not files:
                return [types.TextContent(type="text", text="工作目录为空，还没有任何文件。")]

            file_list = "\n".join(f"  - {f}" for f in sorted(files))
            return [types.TextContent(
                type="text",
                text=f"工作目录（{ALLOWED_ROOT}）中的文件：\n{file_list}"
            )]

    except ValueError as e:
        return [types.TextContent(type="text", text=f"安全错误：{str(e)}")]
    except FileNotFoundError:
        return [types.TextContent(
            type="text",
            text=f"文件不存在：{arguments.get('filename', '')}。请先用 list_files 查看有哪些文件。"
        )]
    except PermissionError:
        return [types.TextContent(type="text", text="权限错误：没有读写该文件的权限")]
    except Exception as e:
        return [types.TextContent(type="text", text=f"操作失败：{str(e)}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream,
                     app.create_initialization_options())

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

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    """列出 Server 提供的所有资源"""
    return [
        types.Resource(
            uri="workspace://readme",           # 资源的唯一标识符
            name="工作区说明",
            description="描述工作区用途和使用规范",
            mimeType="text/plain"               # 内容类型
        ),
        types.Resource(
            uri="workspace://config",
            name="工作区配置",
            description="当前工作区的配置参数",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """读取指定资源的内容"""
    if uri == "workspace://readme":
        return """# AI 工作区说明

这是你的 AI 辅助工作区，位于 ~/Documents/ai-workspace。

## 可用工具
- read_file：读取文件内容
- write_file：写入文件（会覆盖已有内容）
- list_files：查看所有文件

## 使用规范
- 所有文件操作限制在工作区目录内
- 文件名建议使用英文，避免特殊字符
- 重要文件请先备份再修改
"""
    elif uri == "workspace://config":
        config = {
            "allowed_root": ALLOWED_ROOT,
            "max_file_size_kb": 1024,
            "supported_encodings": ["utf-8", "gbk"]
        }
        return json.dumps(config, ensure_ascii=False, indent=2)
    else:
        raise ValueError(f"未知资源：{uri}")

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="code_review",
            description="代码审查提示：读取指定文件并进行详细的代码审查",
            arguments=[
                types.PromptArgument(
                    name="filename",
                    description="要审查的代码文件名（工作目录内的相对路径）",
                    required=True
                ),
                types.PromptArgument(
                    name="language",
                    description="编程语言，如 Python、Java、Go（可选）",
                    required=False
                )
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
    if name == "code_review":
        filename = arguments["filename"]
        language = arguments.get("language", "")

        try:
            path = safe_path(filename)
            with open(path, 'r', encoding='utf-8') as f:
                code = f.read()
        except FileNotFoundError:
            code = f"（文件 {filename} 不存在）"
        except Exception as e:
            code = f"（读取文件失败：{e}）"

        lang_hint = f"（{language}）" if language else ""

        return types.GetPromptResult(
            description=f"审查 {filename} 的代码质量",
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"""请对以下代码{lang_hint}进行专业的代码审查，重点关注：

1. **代码质量**：可读性、命名规范、注释是否充分
2. **潜在问题**：逻辑错误、边界条件、异常处理
3. **性能**：是否有明显的性能问题
4. **安全性**：是否有安全风险（SQL 注入、路径遍历等）
5. **改进建议**：给出具体的、可操作的改进建议

文件：{filename}

请按优先级（高/中/低）整理问题，并提供改进代码示例。"""
                    )
                )
            ]
        )

# npx 是 Node.js 自带的工具，用于运行 npm 包而无需全局安装
# MCP Inspector 是一个可视化调试工具，提供 Web 界面
npx @modelcontextprotocol/inspector python server.py

# 运行 Server，手动发送 JSON-RPC 消息测试
python server.py

{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}