MCP：跨越AI模型与现实的桥梁

curl -LsSf https://astral.sh/uv/install.sh | shuv add "mcp[cli]"

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Simple Server")
@mcp.tool()def add(a: int, b: int) -> int:    """Add两个数字"""    return a + b
@mcp.resource("greeting://{name}")def get_greeting(name: str) -> str:    """获取个性化问候"""    return f"Hello, {name}!"

mcp dev server.py

import sqlite3from fastapi import HTTPExceptionfrom mcp.server.fastmcp import FastMCPimport atexitdb_conn = Nonedef init_db():    global db_conn    db_conn = sqlite3.connect("todo.db")    cursor = db_conn.cursor()    cursor.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, description TEXT NOT NULL, completed INTEGER DEFAULT 0);")    db_conn.commit()init_db()def close_db():    if db_conn:        db_conn.close()atexit.register(close_db)mcp = FastMCP("To-Do List Server")@mcp.tool()def add_task(description: str) -> int:    if not description.strip():        raise ValueError("Task description cannot be empty")    cursor = db_conn.cursor()    cursor.execute("INSERT INTO tasks (description) VALUES (?);", (description,))    task_id = cursor.lastrowid    db_conn.commit()    return task_id@mcp.tool()def mark_task_completed(task_id: int) -> bool:    cursor = db_conn.cursor()    cursor.execute("UPDATE tasks SET completed=1 WHERE id=?;", (task_id,))    rows_affected = cursor.rowcount    db_conn.commit()    return rows_affected > 0@mcp.tool()def delete_task(task_id: int) -> bool:    cursor = db_conn.cursor()    cursor.execute("DELETE FROM tasks WHERE id=?;", (task_id,))    rows_affected = cursor.rowcount    db_conn.commit()    return rows_affected > 0@mcp.tool()def update_task_description(task_id: int, new_description: str) -> bool:    if not new_description.strip():        raise ValueError("Task description cannot be empty")    cursor = db_conn.cursor()    cursor.execute("UPDATE tasks SET description=? WHERE id=?;", (new_description, task_id))    rows_affected = cursor.rowcount    db_conn.commit()    return rows_affected > 0@mcp.resource("tasks://all")def get_all_tasks() -> list[dict]:    cursor = db_conn.cursor()    cursor.execute("SELECT id, description, completed FROM tasks;")    tasks = []    for row in cursor.fetchall():        tasks.append({"id": row[0], "description": row[1], "completed": bool(row[2])})    return tasks@mcp.resource("task_status://{task_id}")def get_task_status(task_id: int) -> dict:    cursor = db_conn.cursor()    cursor.execute("SELECT description, completed FROM tasks WHERE id=?;", (task_id,))    row = cursor.fetchone()    if row is None:        raise HTTPException(status_code=404, detail="Task not found")    return {"description": row[0], "completed": bool(row[1])}@mcp.resource("task_counts://{dummy}")def get_task_counts(dummy: str = "all") -> dict:    cursor = db_conn.cursor()    cursor.execute("SELECT COUNT(*) FROM tasks WHERE completed=0;")    incomplete_count = cursor.fetchone()[0]    cursor.execute("SELECT COUNT(*) FROM tasks WHERE completed=1;")    completed_count = cursor.fetchone()[0]    return {"incomplete": incomplete_count, "completed": completed_count}@mcp.prompt()def todo_list_prompt() -> str:    return """    You have access to a to-do list server. You can use the following tools and resources:    - Tool: add_task(description) -> int      - Adds a new task with the given description and returns the task ID.    - Tool: mark_task_completed(task_id) -> bool      - Marks the task with the given ID as completed and returns True if successful, False otherwise.    - Tool: delete_task(task_id) -> bool      - Deletes the task with the given ID and returns True if successful, False otherwise.    - Tool: update_task_description(task_id, new_description) -> bool      - Updates the description of the task with the given ID and returns True if successful, False otherwise.    - Resource: tasks://all      - Returns a list of all tasks, each with their ID, description, and completion status.    - Resource: task_status://{task_id}      - Returns the status of the task with the given ID, including its description and completion status.    - Resource: task_counts://{dummy}      - Returns a dictionary with the count of incomplete and completed tasks.    You can use these to manage your to-do list.    """

sudo mcp install server.py

2025-03-01T13:15:43.231Z [To-Do List Server] [info] Initializing server...2025-03-01T13:15:43.246Z [To-Do List Server] [error] spawn uv ENOENT {"context":"connection","stack":"Error: spawn uv ENOENT\n    at ChildProcess._handle.onexit (node:internal/child_process:285:19)\n    at onErrorNT (node:internal/child_process:483:16)\n    at process.processTicksAndRejections (node:internal/process/task_queues:82:21)"}2025-03-01T13:15:43.246Z [To-Do List Server] [error] spawn uv ENOENT {"stack":"Error: spawn uv ENOENT\n    at ChildProcess._handle.onexit (node:internal/child_process:285:19)\n    at onErrorNT (node:internal/child_process:483:16)\n    at process.processTicksAndRejections (node:internal/process/task_queues:82:21)"}2025-03-01T13:15:43.248Z [To-Do List Server] [info] Server transport closed2025-03-01T13:15:43.248Z [To-Do List Server] [info] Client transport closed

brew install uv

mcp install server2.py --with fastapi

mcp install server2.py --with fastapi,uvicorn

from fastmcp import FastMCP
mcp = FastMCP("To-Do List Server", dependencies=["fastapi", "uvicorn"])

sudo mcp install server.py

{  "mcpServers": {    "playwright": {      "command": "npx",      "args": ["-y", "@executeautomation/playwright-mcp-server"]    }  }}

#!/usr/bin/env python3"""简单的 MCP 客户端示例连接到 server2.py 并演示 MCP 的基本功能"""import asyncioimport osimport jsonfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom typing import Optional, List, Dict, Any
# 创建连接到 server2.py 的参数server_params = StdioServerParameters(    command="python",     args=[os.path.join(os.path.dirname(os.path.abspath(__file__)), "server2.py")],    env=None  # 使用当前环境变量)
async def run():    """演示 MCP 客户端功能的主函数"""    print("正在连接到 MCP 服务器...")        async with stdio_client(server_params) as (read, write):        async with ClientSession(read, write) as session:            # 1. 初始化连接            await session.initialize()            print("✓ 已成功初始化连接")                        # 2. 获取可用的提示词            try:                prompts = await session.list_prompts()                print(f"可用的提示词: {prompts}")                                if prompts and prompts.prompts:                    # 3. 获取提示词内容                    try:                        prompt = await session.get_prompt(prompts.prompts[0].name)                        print(f"\n服务器提示词:\n{prompt}\n")                    except Exception as e:                        print(f"获取提示词内容时出错: {e}")            except Exception as e:                print(f"获取提示词列表时出错: {e}")                        # 4. 列出可用资源            try:                resources = await session.list_resources()                print(f"可用的资源: {resources}")            except Exception as e:                print(f"获取资源列表时出错: {e}")                resources = []                        # 5. 列出可用工具            try:                tools = await session.list_tools()                print(f"可用的工具: {tools}")            except Exception as e:                print(f"获取工具列表时出错: {e}")                tools = []                    
if __name__ == "__main__":    try:        asyncio.run(run())    except KeyboardInterrupt:        print("\n程序已被用户中断")    except Exception as e:        print(f"\n发生错误: {e}")    finally:        print("演示完成")