邓立国Agent开发入门必读书《AI Agent智能体开发实践》1~11章试读_《ai agent 智能体开发实践》在线阅读-CSDN博客

9.3 单机MCP服务端进阶实现与优化《AI Agent智能体开发实践》-CSDN博客

9.4 单机MCP客户端搭建-CSDN博客

MCP的核心思想是让大语言模型（LLM）能够安全、可控地访问外部工具、数据和计算资源。一个MCP服务器就是一个提供了这些资源和工具的进程，而像Claude、Cursor这样的客户端（AI应用）通过SSE（Server-Sent Events）协议与服务器通信。下面将详细介绍如何使用Python SDK搭建一个单机版MCP服务端。

1. 系统要求

• Python 3.10或更高版本。
• Python MCP SDK 1.2.0或更高版本。
• pip最新版本。
• 推荐使用虚拟环境。

2. 安装Python SDK

pip install mcp-sdk-python

3. 安装必要的依赖库

pip install fastapi uvicorn pydantic python-socketio  # Web框架与数据验证

pip install redis  # 可选，用于上下文存储

4. 核心模块设计

单机MCP服务端主要包含以下核心模块。

• 协议解析模块：处理MCP协议格式。
• 上下文管理模块：存储和管理模型上下文。
• API 接口模块：提供外部访问接口。

5. 一个基础的单机MCP服务端实现代码

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

from typing import Dict, Optional, Any

import uuid

from datetime import datetime, timedelta

import asyncio

# 定义MCP协议数据模型

class ContextRequest(BaseModel):

    model_id: str

    context_data: Dict[str, Any]

    ttl: Optional[int] = 3600  # 上下文过期时间(秒)，默认1小时

class ContextUpdate(BaseModel):

    context_id: str

    context_data: Dict[str, Any]

    append: bool = True  # 是否追加模式，若为False则覆盖

class ContextQuery(BaseModel):

    context_id: str

# 初始化FastAPI应用

app = FastAPI(title="MCP Server (Model Context Protocol)")

# 内存存储上下文数据

class ContextStorage:

    def __init__(self):

        self.contexts: Dict[str, Dict] = {}  # context_id -> {data, expire_time}

    def create_context(self, model_id: str, context_data: Dict, ttl: int) -> str:

        """创建新的上下文"""

        context_id = str(uuid.uuid4())

        expire_time = datetime.now() + timedelta(seconds=ttl)

        self.contexts[context_id] = {

            "model_id": model_id,

            "data": context_data,

            "expire_time": expire_time,

            "created_at": datetime.now()

        }

        return context_id

    def get_context(self, context_id: str) -> Optional[Dict]:

        """获取上下文数据"""

        if context_id not in self.contexts:

            return None

        # 检查是否过期

        context = self.contexts[context_id]

        if datetime.now() > context["expire_time"]:

            del self.contexts[context_id]

            return None

        return context

    def update_context(self, context_id: str, context_data: Dict, append: bool = True) -> bool:

        """更新上下文数据"""

        context = self.get_context(context_id)

        if not context:

            return False

        if append:

            context["data"].update(context_data)

        else:

            context["data"] = context_data

        return True

    def delete_context(self, context_id: str) -> bool:

        """删除上下文"""

        if context_id in self.contexts:

            del self.contexts[context_id]

            return True

        return False

# 初始化上下文存储

context_storage = ContextStorage()

# MCP协议接口实现

@app.post("/mcp/v1/context", response_model=Dict[str, str])

async def create_context(request: ContextRequest):

    """创建新的模型上下文"""

    context_id = context_storage.create_context(

        model_id=request.model_id,

        context_data=request.context_data,

        ttl=request.ttl

    )

    return {"context_id": context_id, "status": "created"}

@app.get("/mcp/v1/context/{context_id}")

async def get_context(context_id: str):

    """获取指定上下文"""

    context = context_storage.get_context(context_id)

    if not context:

        raise HTTPException(status_code=404, detail="Context not found or expired")

    return {

        "context_id": context_id,

        "model_id": context["model_id"],

        "data": context["data"],

        "created_at": context["created_at"]

    }

@app.put("/mcp/v1/context", response_model=Dict[str, str])

async def update_context(update: ContextUpdate):

    """更新上下文数据"""

    success = context_storage.update_context(

        context_id=update.context_id,

        context_data=update.context_data,

        append=update.append

    )

    if not success:

        raise HTTPException(status_code=404, detail="Context not found or expired")

    return {"status": "updated", "context_id": update.context_id}

@app.delete("/mcp/v1/context/{context_id}", response_model=Dict[str, str])

async def delete_context(context_id: str):

    """删除上下文"""

    success = context_storage.delete_context(context_id)

    if not success:

        raise HTTPException(status_code=404, detail="Context not found")

    return {"status": "deleted", "context_id": context_id}

# 启动服务——兼容命令行/Spyder/Jupyter环境

if __name__ == "__main__":

    import uvicorn

    config = uvicorn.Config(app, host="0.0.0.0", port=8000, log_level="info")

    server = uvicorn.Server(config)

    try:

        # 尝试获取当前事件循环（判断是否在 Jupyter/Spyder 等环境中）

        loop = asyncio.get_running_loop()

        print("🟡 检测到已有事件循环（如 Spyder/Jupyter），服务将在后台启动...")

        loop.create_task(server.serve())

        print("✅ MCP Server 正在后台运行中 → http://localhost:8000")

        print("   可访问 API 文档：http://localhost:8000/docs")

        # 注意：在交互式环境中不会阻塞，服务在后台运行

    except RuntimeError:

        # 无事件循环，正常阻塞启动（如命令行直接运行）

        print("🚀 启动 MCP Server...")

        asyncio.run(server.serve())

运行代码，输出如下：

🟡 检测到已有事件循环（如 Spyder/Jupyter），服务将在后台启动...

✅ MCP Server 正在后台运行中 → http://localhost:8000

   可访问 API 文档：http://localhost:8000/docs

INFO:     Started server process [23480]

INFO:     Waiting for application startup.

INFO:     Application startup complete.

INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

6. 服务启动与测试

1）启动MCP服务

python mcp_server.py

#在浏览器中访问主页：http://localhost:8000

2）测试接口（可使用curl或Postman）

（1）创建上下文：

curl -X POST "http://localhost:8000/mcp/v1/context" \

  -H "Content-Type: application/json" \

  -d '{"model_id": "gpt-3.5-turbo", "context_data": {"history": ["user: Hello", "assistant: Hi there!"]}, "ttl": 3600}'

（2）获取上下文：

curl "http://localhost:8000/mcp/v1/context/{context_id}"

（3）更新上下文：

curl -X PUT "http://localhost:8000/mcp/v1/context" \

  -H "Content-Type: application/json" \

  -d '{"context_id": "{context_id}", "context_data": {"history": ["user: How are you?"]}, "append": true}'