Demo Remote MCP Todo

🚀 远程MCP服务器演示（Python）

本指南向您展示了如何构建 远程MCP（模型上下文协议）服务器 从头开始——从简单开始 *你好，世界* 并演变成一个完整的 Todo管理MCP服务器，可通过以下方式访问 超文本传输协议 并可部署到 Azure容器应用.

我们遵循 渐进式、动手型 演示现实世界MCP开发的方法。

______________________________________________________________________

📌 目录

• 第1节——项目设置和基本MCP服务器
• 第2节——实际功能：待办事项管理
• 第3节——MCP工具和HTTP服务器
• 第4节-部署到Azure容器应用程序
• 结束总结
• 资源

______________________________________________________________________

第1节——项目设置和基本MCP服务器

🛠️ 演示命令

1.初始化一个新的Python项目 紫外线

uv init remote-mcp-demo
cd remote-mcp-demo

2.验证设置

uv run hello.py

3.安装Python MCP SDK

uv add "mcp[cli]"

______________________________________________________________________

▶️ 您的第一台MCP服务器

4.创建服务器文件

mkdir src

src/dummy_server.py

from mcp.server.fastmcp import FastMCP

# Create an MCP server using FastMCP
mcp = FastMCP("dummy-mcp-server")

@mcp.tool()
def hello_world(name: str) -> str:
    """Simple hello world tool"""
    return f"Hello, {name}!"

5.测试MCP服务器

uv run mcp dev src/dummy_server.py

______________________________________________________________________

______________________________________________________________________

第2节：构建真正的功能——Todo管理

添加真实的业务逻辑

创建文件： src/db.py

安装Pydantic进行数据验证：

uv add pydantic

db.py –带验证的SQLite数据库层

import sqlite3
import os
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field, field_validator
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("db")

class TodoSchema(BaseModel):
    title: str = Field(..., min_length=1, max_length=255)
    id: Optional[int] = Field(None, gt=0)
    
    @field_validator('title')
    @classmethod
    def validate_title(cls, v: str) -> str:
        if not v or v.strip() == '':
            raise ValueError('Title cannot be empty or contain only whitespace')
        return v.strip()

DB_NAME = "todos.db"

def init_db():
    """Initialize the SQLite database"""
    try:
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS todos (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                text TEXT NOT NULL,
                completed INTEGER NOT NULL DEFAULT 0
            )
        """)
        conn.commit()
        conn.close()
        logger.info(f'Database "{DB_NAME}" initialized.')
    except Exception as error:
        logger.error(f'Error initializing database "{DB_NAME}": {error}')

# Initialize database on import
init_db()

def add_todo(text: str) -> Dict[str, Any]:
    """Add a new todo to the database"""
    logger.info(f"Adding TODO: {text}")

    # Validate input
    validated_input = TodoSchema(title=text)

    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("INSERT INTO todos (text) VALUES (?)", (validated_input.title,))
    lastInsertRowid = cursor.lastrowid
    conn.commit()
    conn.close()
    return {"lastInsertRowid": lastInsertRowid}

def list_todos() -> List[Dict[str, Any]]:
    """Get all todos from the database"""
    logger.info("Listing all TODOs")

    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM todos ORDER BY id")
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]

def complete_todo(todo_id: int) -> Dict[str, Any]:
    """Mark a todo as completed"""
    logger.info(f"Completing TODO: {todo_id}")

    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("UPDATE todos SET completed = 1 WHERE id = ?", (todo_id,))
    changes = cursor.rowcount
    conn.commit()
    conn.close()
    return {"changes": changes}

def delete_todo(todo_id: int) -> Optional[Dict[str, Any]]:
    """Delete a todo from the database"""
    logger.info(f"Deleting TODO: {todo_id}")

    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    # Get the todo before deleting
    cursor.execute("SELECT * FROM todos WHERE id = ?", (todo_id,))
    row = cursor.fetchone()

    if row:
        cursor.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
        conn.commit()
        conn.close()
        return dict(row)

    conn.close()
    return None

def update_todo_text(todo_id: int, text: str) -> Optional[Dict[str, Any]]:
    """Update the text of a todo"""
    logger.info(f"Updating TODO {todo_id}: {text}")

    # Validate input
    validated_input = TodoSchema(title=text, id=todo_id)

    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    cursor.execute("UPDATE todos SET text = ? WHERE id = ?", (validated_input.title, todo_id))

    if cursor.rowcount > 0:
        cursor.execute("SELECT * FROM todos WHERE id = ?", (todo_id,))
        row = cursor.fetchone()
        conn.commit()
        conn.close()
        return dict(row) if row else None

    conn.close()
    return None

______________________________________________________________________

第3节——MCP工具和HTTP服务器

🧰 创建MCP工具

文件： src/tools.py

from mcp.server.fastmcp import FastMCP
from typing import List
from db import (
    add_todo,
    list_todos,
    complete_todo,
    delete_todo,
    update_todo_text,
)

# Create an MCP server
mcp = FastMCP("todo-mcp-server")

@mcp.tool()
def add_todo_tool(title: str) -> List[str]:
    info = add_todo(title)
    return [f"Added TODO: {title} (id: {info['lastInsertRowid']})"]

@mcp.tool()
def list_todos_tool() -> List[str]:
    todos = list_todos()
    if not todos:
        return ["No TODOs found."]
    return [
        f"TODO: {todo['text']} (id: {todo['id']}){' [completed]' if todo['completed'] else ''}"
        for todo in todos
    ]

@mcp.tool()
def complete_todo_tool(id: int) -> List[str]:
    info = complete_todo(id)
    if info["changes"] == 0:
        return [f"TODO with id {id} not found."]
    return [f"TODO with id {id} marked as completed."]

@mcp.tool()
def delete_todo_tool(id: int) -> List[str]:
    row = delete_todo(id)
    if not row:
        return [f"TODO with id {id} not found."]
    return [f"Deleted TODO: {row['text']} (id: {id})"]

@mcp.tool()
def update_todo_text_tool(id: int, text: str) -> List[str]:
    row = update_todo_text(id, text)
    if not row:
        return [f"TODO with id {id} not found."]
    return [f"Updated text for todo with id {id} to \"{text}\""]

测试：

uv run mcp dev src/tools.py

______________________________________________________________________

🧪 使用VS代码副本进行测试

添加到您的VS代码 settings.json:

"my-todo-mcp-server": {
    "type": "stdio",
    "command": "uv",
    "args": [
        "run",
        "mcp",
        "run",
        "/path/to/your/project/src/tools.py"
    ]
}

______________________________________________________________________

🌐 转换为HTTP（远程MCP服务器）

cp src/tools.py src/streamable_http_server.py

在文件底部添加：

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

运行HTTP服务器

uv run python src/streamable_http_server.py

通过检验员进行测试

npx @modelcontextprotocol/inspector http://127.0.0.1:3000/mcp

______________________________________________________________________

第4节-部署到Azure容器应用程序

☁️ 将其投入生产

1.安装Azure Developer CLI

curl -fsSL https://aka.ms/install-azd.sh | bash
curl -fsSL https://aka.ms/install-azd.sh | bash
azd version

2.登录

azd auth login

3.供应+部署

azd init
azd up

注意：确保您有infrarfolder文件、Dockerfile、azure.yaml和azure app_streaable_http_server.py。

______________________________________________________________________

📘 基础设施概述

这 红外线/ 文件夹包含：

• resources.bicep → 容器应用程序、ACR、监控
• main.bicep → 协调部署

测试部署的服务器

npx @modelcontextprotocol/inspector https://mcp-container-py.blacksky-4375de5a.eastus.azurecontainerapps.io/mcp

🧪 将其添加到vscode中

"todo-remote": {
           "type": "http",
           "url": "https://mcp-container-py.blacksky-4375de5a.eastus.azurecontainerapps.io/mcp"
           
    },

______________________________________________________________________

结束总结

✅ 我们建造了什么

• 创建了一个基本 MCP服务器
• 实现 5个Todo CRUD工具
• 转换自 标准输入输出→ HTTP
• 部署到 Azure容器应用 全球可用性

🚀 后续步骤

• 添加更丰富的业务逻辑
• 实现身份验证和RBAC
• 启用速率限制/缓存
• 添加健康检查和监测

______________________________________________________________________

资源

• 演示代码：
• MCP授权深度学习：
• MCP Python SDK：
• MCP规范： https://spec.modelcontextprotocol.io/
• Azure容器应用程序： https://learn.microsoft.com/en-us/azure/container-apps/
• 微软MCP计划：

______________________________________________________________________