MCP协议深度解析：AI Agent生态的USB-C标准正在形成

pip install mcp>=1.0.0

#!/usr/bin/env python3
"""
任务管理 MCP Server
提供任务的增删改查能力
"""

import json
import asyncio
from datetime import datetime
from typing import Optional
from mcp.server import Server
from mcp.types import (
    TextContent,
    Tool,
    Resource,
    Prompt,
    PromptArgument,
)
from mcp.server.stdio import stdio_server

# 内存存储（实际项目应使用数据库）
tasks_db = {}
task_counter = 0

app = Server("task-manager")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """声明Server提供的工具"""
    return [
        Tool(
            name="add_task",
            description="添加新任务",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string", "description": "任务标题"},
                    "description": {"type": "string", "description": "任务描述"},
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                        "description": "优先级"
                    }
                },
                "required": ["title"]
            }
        ),
        Tool(
            name="get_task",
            description="获取任务详情",
            inputSchema={
                "type": "object",
                "properties": {
                    "task_id": {"type": "string", "description": "任务ID"}
                },
                "required": ["task_id"]
            }
        ),
        Tool(
            name="list_tasks",
            description="列出所有任务",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {
                        "type": "string",
                        "enum": ["pending", "completed", "all"],
                        "description": "筛选状态"
                    }
                }
            }
        ),
        Tool(
            name="complete_task",
            description="完成任务",
            inputSchema={
                "type": "object",
                "properties": {
                    "task_id": {"type": "string", "description": "任务ID"}
                },
                "required": ["task_id"]
            }
        ),
        Tool(
            name="delete_task",
            description="删除任务",
            inputSchema={
                "type": "object",
                "properties": {
                    "task_id": {"type": "string", "description": "任务ID"}
                },
                "required": ["task_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """处理工具调用"""
    global task_counter
    
    if name == "add_task":
        task_counter += 1
        task_id = f"task_{task_counter}"
        task = {
            "id": task_id,
            "title": arguments["title"],
            "description": arguments.get("description", ""),
            "priority": arguments.get("priority", "medium"),
            "status": "pending",
            "created_at": datetime.now().isoformat(),
            "completed_at": None
        }
        tasks_db[task_id] = task
        return [TextContent(type="text", text=f"✅ 任务已创建：{task_id}\n{json.dumps(task, indent=2, ensure_ascii=False)}")]
    
    elif name == "get_task":
        task_id = arguments["task_id"]
        if task_id in tasks_db:
            return [TextContent(type="text", text=json.dumps(tasks_db[task_id], indent=2, ensure_ascii=False))]
        return [TextContent(type="text", text=f"❌ 任务不存在：{task_id}")]
    
    elif name == "list_tasks":
        status_filter = arguments.get("status", "all")
        tasks = list(tasks_db.values())
        if status_filter != "all":
            tasks = [t for t in tasks if t["status"] == status_filter]
        
        if not tasks:
            return [TextContent(type="text", text="暂无任务")]
        
        result = f"📋 共 {len(tasks)} 个任务：\n\n"
        for task in tasks:
            status_icon = "✅" if task["status"] == "completed" else "⏳"
            priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}.get(task["priority"], "⚪")
            result += f"{status_icon} [{task['id']}] {priority_icon} {task['title']}\n"
        return [TextContent(type="text", text=result)]
    
    elif name == "complete_task":
        task_id = arguments["task_id"]
        if task_id in tasks_db:
            tasks_db[task_id]["status"] = "completed"
            tasks_db[task_id]["completed_at"] = datetime.now().isoformat()
            return [TextContent(type="text", text=f"✅ 任务已完成：{task_id}")]
        return [TextContent(type="text", text=f"❌ 任务不存在：{task_id}")]
    
    elif name == "delete_task":
        task_id = arguments["task_id"]
        if task_id in tasks_db:
            del tasks_db[task_id]
            return [TextContent(type="text", text=f"🗑️ 任务已删除：{task_id}")]
        return [TextContent(type="text", text=f"❌ 任务不存在：{task_id}")]
    
    return [TextContent(type="text", text=f"未知工具：{name}")]

# Resources：提供只读数据
@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="stats://task-summary",
            name="任务统计",
            mimeType="application/json",
            description="当前任务统计信息"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "stats://task-summary":
        total = len(tasks_db)
        completed = sum(1 for t in tasks_db.values() if t["status"] == "completed")
        pending = total - completed
        return json.dumps({
            "total": total,
            "completed": completed,
            "pending": pending,
            "completion_rate": f"{completed/total*100:.1f}%" if total > 0 else "0%"
        }, ensure_ascii=False)
    raise ValueError(f"未知资源：{uri}")

# Prompts：预定义提示模板
@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="daily_standup",
            description="生成每日站会报告",
            arguments=[
                PromptArgument(
                    name="date",
                    description="报告日期",
                    required=False
                )
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: Optional[dict] = None) -> str:
    if name == "daily_standup":
        date_str = arguments.get("date", datetime.now().strftime("%Y-%m-%d")) if arguments else datetime.now().strftime("%Y-%m-%d")
        completed = [t for t in tasks_db.values() if t["status"] == "completed"]
        pending = [t for t in tasks_db.values() if t["status"] == "pending"]
        
        report = f"# 📅 每日站会报告 ({date_str})\n\n"
        report += f"## ✅ 已完成 ({len(completed)})\n"
        for t in completed:
            report += f"- {t['title']}\n"
        report += f"\n## ⏳ 待处理 ({len(pending)})\n"
        for t in pending:
            report += f"- [{t['priority']}] {t['title']}\n"
        return report
    raise ValueError(f"未知提示：{name}")

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())

#!/usr/bin/env python3
"""
MCP Client 示例
演示如何连接和使用MCP Server
"""

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # 配置Server启动参数
    server_params = StdioServerParameters(
        command="python3",
        args=["task_server.py"],  # 上面创建的Server文件
        env=None
    )
    
    # 建立stdio连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            
            # 1. 列出可用工具
            print("🔧 可用工具：")
            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # 2. 调用工具添加任务
            print("\n➕ 添加任务...")
            result = await session.call_tool("add_task", {
                "title": "完成MCP技术博客",
                "description": "撰写关于MCP协议的深度技术文章",
                "priority": "high"
            })
            print(result.content[0].text)
            
            # 再添加几个任务
            await session.call_tool("add_task", {
                "title": "Review同事代码",
                "priority": "medium"
            })
            await session.call_tool("add_task", {
                "title": "更新项目文档",
                "priority": "low"
            })
            
            # 3. 列出所有任务
            print("\n📋 任务列表：")
            result = await session.list_tools()  # 先确认工具
            tasks = await session.call_tool("list_tasks", {"status": "all"})
            print(tasks.content[0].text)
            
            # 4. 读取资源
            print("\n📊 任务统计：")
            resources = await session.list_resources()
            for resource in resources.resources:
                print(f"  资源: {resource.name} ({resource.uri})")
            
            # 5. 使用Prompt生成报告
            print("\n📝 站会报告：")
            prompts = await session.list_prompts()
            for prompt in prompts.prompts:
                print(f"  提示模板: {prompt.name}")

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

{
  "mcpServers": {
    "task-manager": {
      "command": "python3",
      "args": ["/path/to/task_server.py"],
      "env": {}
    }
  }
}