动态MCP服务器

一个使用FastMCP实现的动态MCP服务器,能够自动从指定目录加载工具、资源和提示模块。

  • by ynu
  • Python
  • 本地部署
  • 更新于

使用说明

MCP-YNU-快速MCP服务器

使用FastMCP的动态MCP服务器实现,可自动从相应目录加载工具、资源和提示。

特性

  • 模块的动态加载 tools/, resources/,以及 prompts/ 目录
  • 模块的自动发现和注册
  • 简单的配置和可扩展性
  • 键入提示以提高代码清晰度和静态分析
  • 用于监控服务器活动的全面日志记录

最近的更新

  • 在整个代码库中添加类型提示
  • 改进了MCP实例处理
  • 添加日志记录功能
  • 已添加MIT许可证
  • 带有参考链接的更新文档

目录结构

mcp-ynu/
├── tools/          # Directory for tool modules
│   ├── __init__.py
│   ├── example.py
├── resources/      # Directory for resource modules
│   ├── __init__.py
│   ├── example.py
├── prompts/        # Directory for prompt modules
│   ├── __init__.py
│   ├── example.py
├── logger.py       # Logger implementation
├── main.py         # Main implementation
├── mcp_server.py   # MCP server implementation
├── README.md       # Project documentation
├── LICENSE         # MIT License
└── pyproject.toml  # Project configuration

用法

  1. 在相应的目录中创建模块
  2. 通过导入mcp from mcp_server import mcp
  3. 运行服务器:
python main.py

示例模块

工具模块示例(Tools/Example.py)

from mcp_server import mcp
import httpx

@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate BMI given weight in kg and height in meters"""
    return weight_kg / (height_m**2)


@mcp.tool()
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
        return response.text

资源模块示例(Resources/Example.py)

from mcp_server import mcp

@mcp.resource("config://app")
def get_config() -> str:
    """Static configuration data"""
    return "App configuration here"


@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Dynamic user data"""
    return f"Profile data for user {user_id}"

提示模块示例(promises/Example.py)

from mcp_server import mcp
from mcp.server.fastmcp.prompts import base

@mcp.prompt()
def review_code(code: str) -> str:
    return f"Please review this code:\n\n{code}"


@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("I'm seeing this error:"),
        base.UserMessage(error),
        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]

调试

  1. 更新 MCP_TRANSPORT_TYPE 在……里面 .env,执行 python main.py 启动mcp服务器
  2. 执行 npx @modelcontextprotocol/inspector 打开 检查.
  3. 选择 SSE 带URL的传输类型 http://localhost:<mcp_server_port>/sse 或选择 STDIO 带命令的运输类型 python 和论点 /path/to/main.py

需求

  • Python>=3.10
  • FastMCP

参考链接

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。