curl -LsSf https://astral.sh/uv/install.sh | sh

uv init my-mcp-server

cd my-mcp-server

uv add "mcp[cli]"

pip install "mcp[cli]"

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("我的第一个MCP服务器")

@mcp.tool()

def add(a: int, b: int) -> int:

    """两数相加"""

    return a + b

if __name__ == "__main__":

    mcp.run()

uv run server.py

@mcp.tool()

def search_users(query: str, limit: int = 10) -> str:

    """搜索用户

    Args:

        query: 搜索关键词

        limit: 返回数量上限

    """

    # 这里调用你的用户搜索逻辑

    results = do_user_search(query, limit)

    return json.dumps(results, ensure_ascii=False)

@mcp.resource("config://app/settings")

def get_app_settings() -> str:

    """获取应用配置"""

    return json.dumps({

        "theme": "dark",

        "language": "zh-CN",

        "version": "1.0.0"

    })

@mcp.resource("users://{user_id}/profile")

def get_user_profile(user_id: str) -> str:

    """获取用户资料"""

    user = db.get_user(user_id)

    return json.dumps(user, ensure_ascii=False)

@mcp.prompt()

def code_review(code: str, language: str = "python") -> str:

    """生成代码审查提示"""

    return f"""请审查以下 {language} 代码，关注：

1. 潜在的 bug

2. 性能问题

3. 代码风格

代码：

```{language}

{code}

```"""

import json

import os

from datetime import datetime

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("任务管理器")

# 数据文件路径

DATA_FILE = os.path.expanduser("~/.task_manager/tasks.json")

def load_tasks() -> list[dict]:

    """加载任务列表"""

    if not os.path.exists(DATA_FILE):

        return []

    with open(DATA_FILE, "r", encoding="utf-8") as f:

        return json.load(f)

def save_tasks(tasks: list[dict]) -> None:

    """保存任务列表"""

    os.makedirs(os.path.dirname(DATA_FILE), exist_ok=True)

    with open(DATA_FILE, "w", encoding="utf-8") as f:

        json.dump(tasks, f, ensure_ascii=False, indent=2)

@mcp.tool()

def add_task(title: str, priority: str = "medium", due_date: str = "") -> str:

    """添加新任务

    Args:

        title: 任务标题

        priority: 优先级，可选 high/medium/low

        due_date: 截止日期，格式 YYYY-MM-DD，可选

    """

    tasks = load_tasks()

    task = {

        "id": len(tasks) + 1,

        "title": title,

        "priority": priority,

        "due_date": due_date,

        "status": "pending",

        "created_at": datetime.now().isoformat()

    }

    tasks.append(task)

    save_tasks(tasks)

    return f"任务已添加：#{task['id']} {title}"

@mcp.tool()

def list_tasks(status: str = "all") -> str:

    """查看任务列表

    Args:

        status: 筛选状态，可选 all/pending/done

    """

    tasks = load_tasks()

    if status != "all":

        tasks = [t for t in tasks if t["status"] == status]

    if not tasks:

        return "没有任务"

    lines = []

    for t in tasks:

        icon = "✅" if t["status"] == "done" else "⬜"

        lines.append(f"{icon} #{t['id']} [{t['priority']}] {t['title']}")

    return "\n".join(lines)

@mcp.tool()

def complete_task(task_id: int) -> str:

    """完成任务

    Args:

        task_id: 任务 ID

    """

    tasks = load_tasks()

    for t in tasks:

        if t["id"] == task_id:

            t["status"] = "done"

            t["completed_at"] = datetime.now().isoformat()

            save_tasks(tasks)

            return f"任务 #{task_id} 已完成：{t['title']}"

    return f"找不到任务 #{task_id}"

@mcp.tool()

def delete_task(task_id: int) -> str:

    """删除任务

    Args:

        task_id: 任务 ID

    """

    tasks = load_tasks()

    original_count = len(tasks)

    tasks = [t for t in tasks if t["id"] != task_id]

    if len(tasks) == original_count:

        return f"找不到任务 #{task_id}"

    save_tasks(tasks)

    return f"任务 #{task_id} 已删除"

@mcp.resource("tasks://stats")

def get_task_stats() -> str:

    """获取任务统计"""

    tasks = load_tasks()

    total = len(tasks)

    done = sum(1 for t in tasks if t["status"] == "done")

    pending = total - done

    high_priority = sum(1 for t in tasks if t["priority"] == "high" and t["status"] == "pending")

    return json.dumps({

        "total": total,

        "done": done,

        "pending": pending,

        "high_priority_pending": high_priority,

        "completion_rate": f"{done/total*100:.1f}%" if total > 0 else "N/A"

    }, ensure_ascii=False)

if __name__ == "__main__":

    mcp.run()

# 修改 server.py 的最后一行

if __name__ == "__main__":

    mcp.run(transport="streamable-http")

npx -y @modelcontextprotocol/inspector

claude mcp add --transport http task-manager http://localhost:8000/mcp

{

  "mcpServers": {

    "task-manager": {

      "command": "uv",

      "args": ["run", "python", "/path/to/server.py"]

    }

  }

}

mcp:

  servers:

    task-manager:

      url: http://localhost:8000/mcp

      transport: streamable-http

mcp:

  servers:

    task-manager:

      command: uv

      args: ["run", "python", "/path/to/server.py"]

      transport: stdio

import sqlite3

from collections.abc import AsyncIterator

from contextlib import asynccontextmanager

from dataclasses import dataclass

from mcp.server.fastmcp import Context, FastMCP

@dataclass

class AppContext:

    db: sqlite3.Connection

@asynccontextmanager

async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:

    """管理数据库连接生命周期"""

    db = sqlite3.connect("tasks.db")

    db.execute("""

        CREATE TABLE IF NOT EXISTS tasks (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            title TEXT NOT NULL,

            priority TEXT DEFAULT 'medium',

            status TEXT DEFAULT 'pending',

            created_at TEXT,

            completed_at TEXT

        )

    """)

    db.commit()

    try:

        yield AppContext(db=db)

    finally:

        db.close()

mcp = FastMCP("任务管理器", lifespan=app_lifespan)

@mcp.tool()

def add_task(title: str, priority: str = "medium", ctx: Context = None) -> str:

    """添加新任务"""

    db = ctx.request_context.lifespan_context.db

    cursor = db.execute(

        "INSERT INTO tasks (title, priority, created_at) VALUES (?, ?, ?)",

        (title, priority, datetime.now().isoformat())

    )

    db.commit()

    return f"任务已添加：#{cursor.lastrowid} {title}"

@mcp.tool()

def list_tasks(status: str = "all", ctx: Context = None) -> str:

    """查看任务列表"""

    db = ctx.request_context.lifespan_context.db

    if status == "all":

        rows = db.execute("SELECT * FROM tasks").fetchall()

    else:

        rows = db.execute("SELECT * FROM tasks WHERE status = ?", (status,)).fetchall()

    if not rows:

        return "没有任务"

    lines = []

    for row in rows:

        icon = "✅" if row[3] == "done" else "⬜"

        lines.append(f"{icon} #{row[0]} [{row[2]}] {row[1]}")

    return "\n".join(lines)

import logging

logging.basicConfig(level=logging.DEBUG)

logger = logging.getLogger("mcp-server")

@mcp.tool()

def my_tool(query: str) -> str:

    logger.info(f"收到请求：{query}")

    try:

        result = do_something(query)

        logger.info(f"返回结果：{result}")

        return result

    except Exception as e:

        logger.error(f"出错了：{e}")

        raise

@mcp.tool()

def risky_operation(path: str) -> str:

    """执行一个可能失败的操作"""

    try:

        result = do_something_risky(path)

        return f"成功：{result}"

    except FileNotFoundError:

        return f"错误：找不到文件 {path}"

    except PermissionError:

        return f"错误：没有权限访问 {path}"

    except Exception as e:

        return f"错误：{str(e)}"

@mcp.tool()

def get_repo_info(owner: str, repo: str) -> str:

    """获取 GitHub 仓库信息"""

    info = github_api.get_repo(owner, repo)

    return json.dumps(info, ensure_ascii=False)  # 必须序列化为字符串

@mcp.tool()

def delete_file(path: str) -> str:

    """删除指定文件"""

    # 必须验证路径！

    if not path.startswith("/safe/directory/"):

        return "错误：只能删除安全目录下的文件"

    if ".." in path:

        return "错误：路径不能包含 .."

    # ... 执行删除

# server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("团队工具服务器")

@mcp.tool()

def team_tool(query: str) -> str:

    """团队共享工具"""

    return f"处理结果：{query}"

if __name__ == "__main__":

    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

# 用 systemd

sudo systemctl enable my-mcp-server

sudo systemctl start my-mcp-server

# 或者用 Docker

docker run -d -p 8000:8000 my-mcp-server

claude mcp add --transport http team-tools http://your-server:8000/mcp