MCP 服务器开发：从本地运行到多用户远程访问

一、为什么需要 MCP 服务器？—— 突破本地开发的局限

前序章节实现的问答机器人仅能在本地运行，面临 “单用户受限、无法远程访问、组件分散管理” 的问题。MCP 服务器的核心价值，是将 MCP 架构的 “三层一储” 组件与业务逻辑集中部署到服务器，实现 “多用户并发访问、组件统一维护、功能远程调用”—— 简单来说，就是让原本只能自己用的问答机器人，变成能让团队、客户通过网络访问的 “公共工具”，同时解决本地开发中 “组件复用难、数据不同步” 的痛点。

MCP 服务器的核心优势可概括为四点：

优势维度	本地开发局限	MCP 服务器解决方案
用户访问	仅单用户本地使用，无法共享	支持多用户远程访问（通过 API 接口），可跨设备（PC / 手机）使用
组件管理	组件分散在本地脚本，更新需逐个修改	组件集中部署在服务器，更新后所有用户即时生效
数据存储	数据保存在本地文件，易丢失	服务器端统一存储数据（支持数据库），数据持久化且可备份
并发处理	无并发能力，仅能单任务运行	支持多用户并发请求（通过异步处理），响应效率更高

二、MCP 服务器技术选型：轻量、高效、易集成

考虑到 MCP 服务器的核心需求是 “快速搭建 API 接口、高效整合组件、支持并发访问”，课程选择以下技术栈，兼顾开发效率与运行性能：

• Web 框架：FastAPI（轻量级、支持异步处理、自动生成 API 文档，适合构建 RESTful API，且与 Python 生态兼容度高，能无缝集成 MCP 组件）；
• 服务器运行器：Uvicorn（异步 ASGI 服务器，专门用于运行 FastAPI 应用，支持高并发，性能优于传统 WSGI 服务器）；
• 请求参数验证：Pydantic（用于定义请求数据模型，自动校验参数格式，避免非法输入导致服务器崩溃）；
• 跨域支持：FastAPI-CORS（解决前端页面与服务器的跨域访问问题，允许指定域名的前端调用 API）；
• 数据存储扩展：SQLite（轻量级数据库，初期用于存储用户会话与文档元数据，后续可无缝切换为 MySQL/PostgreSQL）。

三、MCP 服务器架构设计：组件与 API 的无缝整合

MCP 服务器的架构需围绕 “API 接口为入口、MCP 组件为核心、数据存储为支撑” 设计，将前序的 “三层一储” 组件通过 API 路由暴露给外部用户，同时新增 “请求处理层” 负责参数验证与响应格式化。整体架构如下：

plaintext

[用户/前端] → [API接口层（FastAPI路由）] → [请求处理层（参数验证/响应格式化）] → [MCP核心组件（输入/处理/输出/存储）] → [数据存储层（数据库/文件）]

各层职责与交互逻辑：

1. API 接口层：定义外部可访问的接口（如文档上传、提问、答案导出），指定请求方法（POST/GET）与路径；
2. 请求处理层：用 Pydantic 模型验证请求参数（如文档上传接口需验证文件类型、大小），将合法参数转为 MCP 组件可识别的格式；
3. MCP 核心组件层：复用前序开发的文档解析、LLM 调度、对话存储等组件，按业务逻辑串联执行；
4. 数据存储层：将用户会话、文档元数据、对话历史存入数据库（而非本地文件），支持多用户数据隔离。

四、分步开发：从零搭建 MCP 服务器

1. 环境准备：安装服务器依赖

bash

# 安装FastAPI（Web框架）、Uvicorn（服务器运行器）
pip install fastapi uvicorn
# 安装Pydantic（参数验证）、python-multipart（处理文件上传）
pip install pydantic python-multipart
# 安装CORS扩展（解决跨域）、SQLAlchemy（数据库ORM，可选）
pip install fastapi-cors sqlalchemy

2. 服务器初始化：整合 MCP 组件与 FastAPI

首先创建服务器主文件（mcp_server.py），完成 FastAPI 应用初始化、CORS 配置、MCP 组件集成：

python

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel  # 用于参数验证
import os
# 导入前序开发的MCP组件
from mcp_core.input import DocumentParser, TextInputCleaner
from mcp_core.process import LLMCaller, ContextManager
from mcp_core.output import TextFormatter, DocExporter
from mcp_core.storage import ChatStorage, DocMetadataStorage

# ---------------------- 1. 初始化FastAPI应用 ----------------------
app = FastAPI(title="MCP Server", version="1.0")

# ---------------------- 2. 配置CORS（允许跨域访问） ----------------------
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 开发阶段允许所有域名，生产环境需指定具体域名（如"https://your-frontend.com"）
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有HTTP方法（GET/POST/PUT等）
    allow_headers=["*"],  # 允许所有请求头
)

# ---------------------- 3. 初始化MCP组件（复用前序逻辑，适配服务器场景） ----------------------
# 输入层组件：指定服务器端文件存储路径
UPLOAD_DIR = "./server_uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)  # 确保上传目录存在
doc_parser = DocumentParser(supported_formats=["pdf", "docx"], allow_encrypted=True)
text_cleaner = TextInputCleaner(max_length=500, filter_sensitive=True)

# 处理层组件：启用组件池（复用LLM连接，避免重复初始化）
llm_caller = LLMCaller(model="gpt-4o", api_key="your-openai-api-key", params={"temperature": 0.2})
context_manager = ContextManager(max_history=5)

# 输出层组件：指定服务器端导出目录
EXPORT_DIR = "./server_exports"
os.makedirs(EXPORT_DIR, exist_ok=True)
text_formatter = TextFormatter(
    format_template="【用户问题】{user_question}\n【回答内容】{answer}\n【参考文档】{doc_name}（第{page_num}页）"
)
doc_exporter = DocExporter(export_format="markdown", save_dir=EXPORT_DIR, add_timestamp=True)

# 存储层组件：切换为数据库存储（示例用SQLite，替代本地JSON文件）
chat_storage = ChatStorage(storage_type="db", db_path="mcp_server.db")  # 数据库路径
doc_meta_storage = DocMetadataStorage(storage_type="db", db_path="mcp_server.db")

3. 定义请求模型：用 Pydantic 验证参数

服务器需严格验证用户输入（如文件类型、会话 ID 格式），避免非法请求导致组件报错。通过 Pydantic 定义请求模型：

python

# 定义提问请求模型（验证session_id、doc_name、user_question）
class QuestionRequest(BaseModel):
    session_id: str  # 区分多用户的会话ID（如"user_123456"）
    doc_name: str    # 目标文档名称（需已上传）
    user_question: str  # 用户提问内容

# 定义导出请求模型（验证session_id）
class ExportRequest(BaseModel):
    session_id: str

4. 开发 API 路由：映射业务功能

按 “文档上传→提问→答案导出” 的业务流程，开发对应的 API 接口：

路由 1：文档上传（POST /api/upload）

负责接收用户上传的文档，调用 MCP 输入层组件解析，并存入服务器目录：

python

@app.post("/api/upload")
async def upload_document(
    session_id: str,  # 从请求参数获取会话ID
    file: UploadFile = File(...)  # 接收上传的文件
):
    # 1. 验证文件类型（仅允许PDF/Word）
    allowed_extensions = {"pdf", "docx"}
    file_ext = file.filename.split(".")[-1].lower()
    if file_ext not in allowed_extensions:
        raise HTTPException(status_code=400, detail=f"不支持的文件类型，仅允许{allowed_extensions}")
    
    # 2. 保存文件到服务器上传目录
    file_path = os.path.join(UPLOAD_DIR, f"{session_id}_{file.filename}")  # 用session_id区分用户文件
    with open(file_path, "wb") as f:
        f.write(await file.read())  # 异步读取文件（适配FastAPI异步特性）
    
    # 3. 调用MCP文档解析组件处理
    parse_result = doc_parser.parse(file_path=file_path)
    if parse_result["status"] == "failed":
        raise HTTPException(status_code=500, detail=f"文档解析失败：{parse_result['error']}")
    
    # 4. 存储文档元数据（关联session_id）
    doc_meta = {
        "session_id": session_id,
        "doc_name": file.filename,
        "file_path": file_path,
        "page_count": parse_result["data"]["page_count"],
        "upload_time": parse_result["data"]["parse_time"]
    }
    doc_meta_storage.save(metadata=doc_meta)
    
    # 5. 返回成功响应
    return {
        "status": "success",
        "data": {
            "doc_name": file.filename,
            "page_count": parse_result["data"]["page_count"],
            "message": "文档上传并解析成功"
        },
        "error": None
    }

路由 2：用户提问（POST /api/answer）

接收用户提问，调用 MCP 处理层组件生成答案，关联用户会话：

python

@app.post("/api/answer")
def get_answer(request: QuestionRequest):
    # 1. 验证请求参数（Pydantic自动验证，若不合法直接返回422错误）
    session_id = request.session_id
    doc_name = request.doc_name
    user_question = request.user_question
    
    # 2. 清洗用户提问（调用输入层组件）
    clean_result = text_cleaner.clean(input_text=user_question)
    if not clean_result["data"]["is_valid"]:
        raise HTTPException(status_code=400, detail=f"提问无效：{clean_result['error']}")
    cleaned_question = clean_result["data"]["cleaned_text"]
    
    # 3. 获取用户关联的文档（从存储层筛选session_id对应的文档）
    doc_meta_list = doc_meta_storage.get_by_session(session_id=session_id)
    target_doc = next((doc for doc in doc_meta_list if doc["doc_name"] == doc_name), None)
    if not target_doc:
        raise HTTPException(status_code=404, detail=f"未找到用户{session_id}的文档：{doc_name}")
    
    # 4. 调用LLM生成答案（处理层组件）
    doc_content = doc_parser.get_parsed_text(file_path=target_doc["file_path"])
    history = context_manager.get_history(session_id=session_id)
    prompt = f"基于文档内容回答：{doc_content}\n历史对话：{history}\n当前问题：{cleaned_question}"
    
    llm_result = llm_caller.generate(prompt=prompt)
    if llm_result["status"] == "failed":
        raise HTTPException(status_code=500, detail=f"答案生成失败：{llm_result['error']}")
    raw_answer = llm_result["data"]["output_text"]
    
    # 5. 格式化答案（输出层组件）
    formatted_answer = text_formatter.format(
        user_question=cleaned_question,
        answer=raw_answer,
        doc_name=doc_name,
        page_num=llm_result["data"]["reference_page"]
    )
    
    # 6. 保存对话记录（关联session_id）
    chat_record_user = {
        "session_id": session_id,
        "role": "user",
        "content": cleaned_question,
        "timestamp": llm_result["data"]["generate_time"]
    }
    chat_record_ai = {
        "session_id": session_id,
        "role": "ai",
        "content": formatted_answer,
        "timestamp": llm_result["data"]["generate_time"]
    }
    chat_storage.save(record=chat_record_user)
    chat_storage.save(record=chat_record_ai)
    
    # 7. 更新上下文
    context_manager.update_history(session_id=session_id, user_msg=cleaned_question, ai_msg=formatted_answer)
    
    # 8. 返回答案
    return {
        "status": "success",
        "data": {"formatted_answer": formatted_answer},
        "error": None
    }

路由 3：答案导出（POST /api/export）

导出用户的对话历史，生成可下载的文档：

python

@app.post("/api/export")
def export_chat(request: ExportRequest):
    session_id = request.session_id
    
    # 1. 获取用户对话历史
    chat_history = chat_storage.get_by_session(session_id=session_id)
    if not chat_history:
        raise HTTPException(status_code=404, detail=f"用户{session_id}无对话记录可导出")
    
    # 2. 调用输出层组件导出
    export_result = doc_exporter.export(
        content=chat_history,
        file_prefix=f"chat_{session_id}"  # 文件名包含session_id，避免冲突
    )
    if export_result["status"] == "failed":
        raise HTTPException(status_code=500, detail=f"导出失败：{export_result['error']}")
    
    # 3. 返回导出文件路径（前端可通过该路径下载）
    return {
        "status": "success",
        "data": {
            "file_path": export_result["data"]["file_path"],
            "file_name": export_result["data"]["file_name"],
            "download_url": f"/exports/{export_result['data']['file_name']}"  # 前端下载URL
        },
        "error": None
    }

路由 4：静态文件访问（GET /exports/{file_name}）

用于前端下载导出的文档，FastAPI 通过StaticFiles挂载静态目录：

python

from fastapi.staticfiles import StaticFiles

# 挂载导出目录为静态文件目录，允许外部访问
app.mount("/exports", StaticFiles(directory=EXPORT_DIR), name="exports")

5. 多用户支持的关键设计

• 会话隔离：通过session_id（用户登录后生成或前端随机生成）关联所有用户数据（文档、对话、导出文件），避免不同用户数据混淆；
• 文件隔离：用户上传的文件命名格式为{session_id}_{原始文件名}，存储在统一目录下，既集中管理又避免文件名冲突；
• 组件池复用：LLM 调度组件、文档解析组件在服务器启动时初始化一次，后续请求直接复用，避免每次请求重新创建组件（减少 API 调用延迟与资源消耗）。

五、服务器部署与接口测试

1. 启动服务器

通过 Uvicorn 启动 FastAPI 应用，支持异步处理与热重载（开发阶段）：

bash

# 命令格式：uvicorn 文件名:应用实例 --host 主机地址 --port 端口号 --reload（热重载，开发用）
uvicorn mcp_server:app --host 0.0.0.0 --port 8000 --reload

• --host 0.0.0.0：允许外部设备访问（而非仅本地127.0.0.1）；
• --port 8000：指定服务器端口（若 8000 被占用，可改为 8080、9000 等）；
• --reload：代码修改后自动重启服务器，适合开发阶段。

启动成功后，访问http://服务器IP:8000/docs可查看 FastAPI 自动生成的 API 文档（支持在线调试接口）。

2. 接口测试（以 Postman 为例）

测试文档上传接口（POST /api/upload）

• 请求参数：session_id（如user_001）作为 Query 参数，file选择本地 PDF/Word 文件作为 Form 数据；
• 预期响应：返回status: success，包含文档名称与页数。

测试提问接口（POST /api/answer）

• 请求体（JSON 格式）：

  json

  {
    "session_id": "user_001",
    "doc_name": "test.pdf",
    "user_question": "文档中MCP架构的核心分层有哪些？"
  }
  

• 预期响应：返回格式化的答案。

测试导出接口（POST /api/export）

• 请求体（JSON 格式）：

  json

  {
    "session_id": "user_001"
  }
  

• 预期响应：返回下载 URL，访问该 URL 可获取 Markdown 格式的对话记录。

六、常见问题与解决方案

问题类型	具体表现	解决方案
跨域访问错误	前端调用接口时提示 “Access-Control-Allow-Origin” 错误	检查 CORS 配置，确保allow_origins包含前端域名（生产环境避免用"*"）
端口占用	启动服务器时提示 “Address already in use”	更换端口（如--port 8080），或关闭占用端口的进程（lsof -i:8000查看进程，kill -9 进程ID关闭）
文档上传失败	提示 “文件过大” 或 “读取失败”	1. 在UploadFile处理中添加文件大小限制（如if file.size > 10*1024*1024: 拒绝上传）；2. 确保服务器目录有写入权限
LLM 并发错误	多用户同时提问时提示 “API 请求超时”	1. 给 LLM 组件添加请求队列（如用asyncio.Queue）；2. 降低temperature等参数减少 LLM 计算时间；3. 切换为支持更高并发的 LLM 服务

七、总结与后续扩展方向

1. 本集核心收获

• 掌握 MCP 服务器的核心价值：从 “本地单用户” 升级为 “多用户远程访问”，实现组件集中管理与数据持久化；
• 学会用 FastAPI 整合 MCP 组件，开发标准化 API 接口，解决跨域、参数验证、多用户隔离等服务器关键问题；
• 理解服务器部署与测试流程，能独立启动 MCP 服务器并调试接口。

2. 后续扩展方向

• 容器化部署：用 Docker 打包 MCP 服务器（包含依赖、组件、配置），实现 “一次打包，多环境运行”；
• 用户认证：集成 OAuth2.0 或 JWT 实现用户登录认证，替代当前的session_id随机生成，提升安全性；
• 性能监控：添加 Prometheus+Grafana 监控服务器 CPU、内存、接口响应时间，及时发现性能瓶颈；
• 容错与备份：实现组件故障自动恢复（如 LLM 组件调用失败时切换备用模型）、数据定时备份，提升服务器可用性。