MCP项目实战-基于FastMCP

zh_erhao

438人浏览 · 2025-12-12 09:00:00

zh_erhao · 2025-12-12 09:00:00 发布

之前的一个测试实战项目，开源出来供大家参考：

基于 FastMCP 框架构建的 Model Context Protocol (MCP) 服务器，集成了 SQLite 数据库和现代化的 FastAPI 网页管理界面。支持 SSE/stdio 协议，适合中文场景的知识与配置管理。

主要特性

MCP 协议支持：兼容 Claude、ChatGPT 等 AI 客户端，支持 create/update/delete/list/search 等标准工具调用。
Web 可视化管理：内置 FastAPI 管理后台（自动并发启动，端口 8000），可视化增删改查和搜索。
数据项类型/状态全面中文化：所有类型与状态枚举均为中文，支持从标题自动提取类型。
丰富的 API 工具与资源：支持 RESTful 工具、资源 URI、提示模板等，便于二次开发和自动化。
极简部署：仅需 Python 3.10+，pip 安装依赖即可运行。
详细文档：docs/ 目录下包含 API、数据库、配置等详细说明。

安装与运行

pip install fastmcp sqlalchemy
python3 main.py --transport sse --host 0.0.0.0 --port 8008

MCP 主服务端口由 config.py 配置（默认 8008）
Web 管理界面自动运行在 http://localhost:8000

目录结构

/mcpserver
  ├── main.py            # 主程序入口，自动并发启动MCP服务与Web管理界面
  ├── web_interface.py   # FastAPI网页管理界面
  ├── config.py          # 配置项
  ├── database.py        # 数据库模型与操作
  ├── templates/         # 网页模板
  ├── static/            # 静态文件
  ├── data.db            # SQLite数据库
  ├── README.md
  └── docs/              # 详细文档（API、配置、数据库等）

快速体验

启动服务后，浏览器访问 http://localhost:8000 进入可视化管理后台。
使用 Claude/ChatGPT 等 AI 客户端可直接调用 MCP 工具。
支持数据项类型/状态自动中文识别，创建/编辑更智能。

典型工具与资源

create_item_tool：创建数据项，类型自动提取
update_item_tool：更新数据项，支持类型自动识别
delete_item_tool：删除数据项
list_items_tool：列出所有启用数据项
search_items_by_title：按标题模糊搜索
db://item/{item_id}：获取单项资源
db://items：获取全部资源列表
item_template/list_template：格式化输出模板

核心代码：

"""
FastMCP服务器入口模块 - 主应用程序
使用FastMCP框架创建MCP服务器，集成SQLite数据库
支持 stdio 和 sse 两种传输方式
"""

import argparse
import re
from fastmcp import FastMCP
from database import (
    init_db, create_item, get_item, get_all_items, update_item, delete_item,
    SessionLocal, Item, ItemType, ItemStatus
)
from typing import Optional, List, Dict, Any, Literal, Tuple
from config import SERVER_TITLE, SERVER_DESCRIPTION, RESOURCE_PREFIX, SERVER_HOST, SERVER_PORT, ENABLE_SSE


def extract_item_type_from_title(title: str) -> Tuple[str, str]:
    """
    从标题中提取类型信息
    
    参数:
        title: 标题文本
        
    返回:
        Tuple[处理后的标题, 提取的类型]
    """
    # 定义类型关键词映射
    type_patterns = {
        r'^\[?(规范|标准)\]?[:：]\s*': ItemType.规范,
        r'^\[?(接口|API)\]?[:：]\s*': ItemType.接口,
        r'^\[?(枚举|ENUM)\]?[:：]\s*': ItemType.枚举,
        r'^\[?(字段|FIELD)\]?[:：]\s*': ItemType.字段,
        r'^\[?(示例|EXAMPLE)\]?[:：]\s*': ItemType.示例,
        r'^\[?(说明|DESC|DESCRIPTION)\]?[:：]\s*': ItemType.说明
    }
    
    # 尝试匹配类型
    for pattern, item_type in type_patterns.items():
        if re.match(pattern, title, re.IGNORECASE):
            # 移除类型前缀并清理标题
            clean_title = re.sub(pattern, '', title, flags=re.IGNORECASE).strip()
            return clean_title, item_type
    
    # 如果没有匹配到类型，返回原标题和默认类型
    return title, ItemType.其他

# 创建FastMCP服务器 (避免传递运行时和传输特定设置作为kwargs)
mcp = FastMCP(
    name=SERVER_TITLE
    # 移除了运行时和传输特定设置，这些将在run()方法中提供
    # 移除了不支持的version参数
)

# 定义工具
@mcp.tool()
def create_item_tool(
    title: str, 
    content: str,
    item_type: Optional[str] = None
) -> Dict[str, Any]:
    """
    创建一个新的数据项
    
    参数:
        title: 数据项标题
        content: 数据项内容
        item_type: 数据项类型（可选），如果未指定则从标题中提取
             可选值：规范, 接口, 枚举, 字段, 示例, 说明, 其他
        
    返回:
        创建的数据项信息
    """
    try:
        # 如果未指定类型，尝试从标题中提取
        clean_title = title
        if not item_type:
            clean_title, extracted_type = extract_item_type_from_title(title)
            item_type_enum = extracted_type
        else:
            try:
                item_type_enum = ItemType(item_type)
            except ValueError:
                return {"error": f"无效的类型: {item_type}"}
        
        item = create_item(
            title=clean_title, 
            content=content,
            item_type=item_type_enum,
            status=ItemStatus.启用  # 默认启用
        )
        return item.to_dict()
    except Exception as e:
        return {"error": f"创建数据项失败: {str(e)}"}


@mcp.tool()
def update_item_tool(
    item_id: int, 
    title: Optional[str] = None, 
    content: Optional[str] = None,
    item_type: Optional[str] = None
) -> Dict[str, Any]:
    """
    更新现有数据项的内容
    
    参数:
        item_id: 要更新的数据项ID (必填)
        title: 新的数据项标题（可选）
        content: 新的数据项内容（可选）
        item_type: 新的数据项类型（可选）
             可选值：规范, 接口, 枚举, 字段, 示例, 说明, 其他
        
    返回:
        更新后的数据项信息或错误信息
    """
    try:
        # 验证并转换类型
        item_type_enum = None
        clean_title = title
        
        if item_type:
            try:
                item_type_enum = ItemType(item_type)
            except ValueError:
                return {"error": f"无效的类型: {item_type}"}
        elif title:  # 如果提供了新标题但未提供类型，尝试从标题中提取
            clean_title, extracted_type = extract_item_type_from_title(title)
            item_type_enum = extracted_type
        
        result = update_item(
            item_id=item_id,
            title=clean_title if clean_title else title,  # 使用清理后的标题（如果可用）
            content=content,
            item_type=item_type_enum,
            status=None  # 不更新状态
        )
        if result:
            return result.to_dict()
        return {"error": "项目不存在或已被禁用"}
    except Exception as e:
        return {"error": f"更新数据项失败: {str(e)}"}


@mcp.tool()
def delete_item_tool(item_id: int) -> Dict[str, Any]:
    """
    删除指定的数据项
    
    参数:
        item_id: 要删除的数据项ID
        
    返回:
        操作结果
    """
    success = delete_item(item_id)
    if success:
        return {"success": True}
    return {"error": "项目不存在"}


@mcp.tool()
def list_items_tool() -> Dict[str, List[Dict[str, Any]]]:
    """
    列出所有启用的数据项
    
    返回:
        所有启用的数据项列表
    """
    items = get_all_items(include_disabled=False)
    return {"items": [item.to_dict() for item in items]}


@mcp.tool()
def list_item_titles() -> Dict[str, List[Dict[str, Any]]]:
    """
    列出所有数据项的标题和ID
    
    返回:
        包含所有数据项标题和ID的列表
        
        示例返回:
        {
            "items": [
                {"id": 1, "title": "示例标题1"},
                {"id": 2, "title": "示例标题2"}
            ]
        }
    """
    items = get_all_items()
    return {
        "items": [
            {"id": item.id, "title": item.title}
            for item in items
        ]
    }


@mcp.tool()
def search_items_by_title(
    title: str, 
    item_type: Optional[str] = None
) -> Dict[str, List[Dict[str, Any]]]:
    """
    根据标题搜索启用的数据项（不区分大小写的模糊匹配）
    
    参数:
        title: 要搜索的标题关键词
        item_type: 筛选特定类型的数据项（可选）
             可选值：规范, 接口, 枚举, 字段, 示例, 说明, 其他
        
    返回:
        匹配的启用数据项列表
        
        示例返回:
        {
            "items": [
                {
                    "id": 1,
                    "title": "示例标题",
                    "content": "这是内容...",
                    "type": "规范",
                    "status": "启用",
                    "created_at": "2023-01-01T00:00:00",
                    "updated_at": "2023-01-01T00:00:00"
                }
            ]
        }
    """
    db = SessionLocal()
    try:
        # 构建查询，默认只查询启用的项
        query = db.query(Item).filter(
            Item.title.ilike(f'%{title}%'),
            Item.status == ItemStatus.启用
        )
        
        # 添加类型筛选
        if item_type:
            try:
                item_type_enum = ItemType(item_type)
                query = query.filter(Item.type == item_type_enum)
            except ValueError:
                return {"error": f"无效的类型: {item_type}"}
        
        # 执行查询并返回结果
        items = query.all()
        return {"items": [item.to_dict() for item in items]}
    except Exception as e:
        db.rollback()
        return {"error": f"搜索数据项失败: {str(e)}"}
    finally:
        db.close()


# 定义资源
@mcp.resource(f"{RESOURCE_PREFIX}item/{{item_id}}")
def item_resource(item_id: str) -> Dict[str, Any]:
    """
    获取指定ID的数据项
    
    参数:
        item_id: 数据项ID
        
    返回:
        数据项信息或错误信息
    """
    item = get_item(int(item_id))
    if item:
        return item.to_dict()
    return {"error": "项目不存在"}


@mcp.resource(f"{RESOURCE_PREFIX}items")
def items_resource() -> Dict[str, List[Dict[str, Any]]]:
    """
    获取所有数据项的列表
    
    返回:
        所有数据项的列表
    """
    items = get_all_items()
    return {"items": [item.to_dict() for item in items]}


# 定义提示模板
@mcp.prompt("item_template")
def item_template(data: Dict[str, Any]) -> str:
    """
    格式化显示单个数据项
    
    参数:
        data: 数据项信息
        
    返回:
        格式化后的数据项显示
    """
    return f"""
# 数据项: {data['title']}

**ID**: {data['id']}
**创建时间**: {data['created_at']}
**更新时间**: {data['updated_at']}

## 内容

{data['content']}
"""


@mcp.prompt("list_template")
def list_template(data: Dict[str, List[Dict[str, Any]]]) -> str:
    """
    格式化显示数据项列表
    
    参数:
        data: 包含数据项列表的字典
        
    返回:
        格式化后的数据项列表显示
    """
    if not data.get("items") or len(data["items"]) == 0:
        return "# 数据项列表\n\n当前没有数据项。"
        
    items_text = "\n\n".join([
        f"## {item['title']}\n\n**ID**: {item['id']}\n**创建时间**: {item['created_at']}\n**更新时间**: {item['updated_at']}"
        for item in data["items"]
    ])
    
    return f"""
# 数据项列表

共找到 {len(data['items'])} 个数据项:

{items_text}
"""


def parse_args():
    """
    解析命令行参数
    
    返回:
        argparse.Namespace: 解析后的命令行参数
    """
    parser = argparse.ArgumentParser(description='启动 FastMCP 服务器')
    parser.add_argument('--transport', 
                        type=str, 
                        choices=['stdio', 'sse'], 
                        default='stdio',
                        help='传输方式: stdio 或 sse (默认: stdio)')
    parser.add_argument('--host', 
                        type=str, 
                        default=SERVER_HOST,
                        help=f'服务器主机地址 (默认: {SERVER_HOST})')
    parser.add_argument('--port', 
                        type=int, 
                        default=SERVER_PORT,
                        help=f'服务器端口 (默认: {SERVER_PORT})')
    parser.add_argument('--log-level', 
                        type=str, 
                        choices=['debug', 'info', 'warning', 'error', 'critical'], 
                        default='info',
                        help='日志级别 (默认: info)')
    parser.add_argument('--debug', 
                        action='store_true',
                        help='启用调试模式')
    return parser.parse_args()

import threading  # 新增：用于并发启动web服务
import uvicorn    # 新增：用于运行FastAPI应用
import time      # 添加时间模块，用于延迟处理

def main():
    """
    主函数：初始化数据库并启动服务器，并自动启动Web管理界面（FastAPI）
    处理初始化顺序并确保服务器组件正确启动
    """
    # 初始化数据库
    init_db()
    
    # 解析命令行参数
    args = parse_args()

    # 启动 web_interface 的 FastAPI 应用（以新线程方式）
    def run_web_interface():
        """
        启动 web_interface.py 中的 FastAPI 应用
        """
        try:
            from web_interface import app
            # 增加启动延迟，确保主线程有时间初始化
            time.sleep(0.5)
            print("正在启动Web界面...")
            # 测试端口8000是否被占用
            import socket
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                s.bind(("0.0.0.0", 8000))
                s.close()
                # 端口可用
                uvicorn.run(
                    app,
                    host="0.0.0.0",
                    port=8000,
                    log_level="info",
                    timeout_keep_alive=120,
                    access_log=False  # 关闭访问日志，减少并发冲突
                )
            except OSError:
                # 端口被占用，尝试使用其他端口
                print("Web界面端口8000已被占用，尝试端口8001...")
                try:
                    uvicorn.run(
                        app,
                        host="0.0.0.0",
                        port=8001,
                        log_level="info",
                        timeout_keep_alive=120,
                        access_log=False
                    )
                except Exception as e:
                    print(f"Web界面启动失败: {e}")
            finally:
                if not s.closed:
                    s.close()
        except Exception as e:
            print(f"Web界面启动错误: {e}")
            import traceback
            traceback.print_exc()
            
    web_thread = threading.Thread(target=run_web_interface, daemon=True)
    web_thread.start()
    print("Web 管理界面启动中...")
    
    # 添加短暂延迟，确保各组件有足够时间初始化
    time.sleep(1.5)  # 增加等待时间
    print("Web 管理界面已启动：http://localhost:8000")

    # 启动主MCP服务器
    if __name__ == "__main__":
        if args.transport == 'sse':
            # 使用 SSE 传输方式
            print(f"启动 SSE 服务器在 {args.host}:{args.port}/sse")
            try:
                # 重构SSE初始化流程
                time.sleep(3.0)  # 先等待Web界面启动
                print("等待服务器初始化...")
                
                # 使用uvicorn_config配置服务器的待机和超时时间
                # 简化uvicorn配置，只使用最必要的参数
                uvicorn_config = {
                    "timeout_keep_alive": 120,  # 增加keep-alive超时时间
                }
                
                # 根据调试模式设置日志级别
                log_level = "debug" if args.debug else args.log_level
                
                # 使用支持的参数启动SSE服务器
                print(f"使用日志级别：{log_level}")
                mcp.run(
                    transport="sse",
                    host=args.host,
                    port=args.port,
                    path="/sse",
                    log_level=log_level
                    # 去除可能导致兼容性问题的参数
                )
            except Exception as e:
                print(f"SSE服务器启动错误: {e}")
                import traceback
                traceback.print_exc()
                print("\n错误分析: 此错误可能是由于ASGI协议处理异常导致的")
                print("\n详细错误信息:")
                import traceback
                traceback.print_exc()
                print("\n解决方案: ")
                print("1. 尝试使用stdio模式: python3 main.py --transport stdio")
                print("2. 如需使用SSE模式，请确保版本兼容: python3 -m pip install -U fastmcp mcp")
                print("3. 检查是否有其他服务占用端口: lsof -i :8008")
        else:
            # 使用 Stdio 传输方式（默认）
            print("启动 Stdio 服务器...")
            try:
                # stdio模式不支持log_level参数，需要移除
                mcp.run(
                    transport="stdio"
                    # stdio模式不传递log_level参数
                )
            except Exception as e:
                print(f"Stdio服务器启动错误: {e}")
                import traceback
                traceback.print_exc()


# 当直接运行此文件时，启动服务器
if __name__ == "__main__":
    main()

项目地址：https://gitee.com/zhangerhao/mcpmanager.git