使用说明
MCP Core(核心)
MCP核心 它是Python中一个模块化控制平台(MCP)框架的核心模块,旨在运行一个支持通过装饰器扩展工具的FastMCP服务器。
它允许集中管理逻辑、配置、日志记录和数据库访问,而将实现特定功能的任务留给工具来完成。
项目结构
./
├── .gitignore
├── pyproject.toml
├── readme.md
└── src/
└── mcp_core/
├── __init__.py
├── cli.py
├── config.py
├── core.py
├── helpers/
│ ├── logger.py
│ ├── parquet.py
│ └── sqlite.py
├── loader.py
└── server.py
运作/功能
CoreAPI(核心应用程序编程接口)
CoreAPI 它是项目的核心内容,包括:
config主要配置(主机、端口、路径、协议等)logger通过集中式日志记录LoggerHelpermcp申请/请求FastMCP用于注册和运行工具
所有工具都可以访问 core_api 全球通过:
from mcp_core.core import core_api
工具
I tools(工具)是位于tools/文件夹中的Python模块。每个需要注册为工具的函数都会被装饰器标记如下:
@core_api.mcp.tool
def health_check() -> dict:
return {"status": "ok"}
加载器自动导入了tools文件夹中的所有模块(包括递归导入),并注册了被装饰器修饰的函数。 无需从加载器中显式导出或返回函数。
装载器
loader.py 负责:
- 动态导入工具文件夹中的所有Python模块
- 管理物理路径(/app/tools)和Python包(工具)
- 自动注册带有 @core_api.mcp.tool 注解的工具
使用示例:
from mcp_core.loader import load_tools
tools = load_tools("/app/tools") # o "tools" se è un package Python
服务器
server.py 管理MCP服务器的启动。它可以接收:
- 连接的主机、端口、路径
- 由加载器加载的工具
- 可选的start_callback用于在启动时执行自定义逻辑
CLI(Command Line Interface,命令行界面)
cli.py 它是启动服务器的主要入口点。
主要功能:
- 从CoreConfig读取配置
- 通过 load_tools 加载工具
- 使用加载的工具实例化FastMCP
- 启动带有自定义回调的MCP服务器
助手们
它们应被视为辅助插件,用于整合各种给定数据源或工具(目前未按功能划分,后续可根据需要进行评估)。 目前我们有:
logger.py集中式日志记录的封装器(LoggerHelper)sqlite.py用于管理多个SQLite数据库、查询和Pandas数据框的辅助工具parquet.py辅助程序用于读写Parquet文件
配置
主要配置由CoreConfig读取,包括:
- 实例名称:服务器名称
- 主机,端口,路径:监听端点
- 协议:传输类型(http)
- log_level: 日志级别
使用示例:
from mcp_core.core import core_api
logger = core_api.logger
config = core_api.config
logger.info(f"Server '{config.core.get('instance_name')}' pronto all'avvio")
服务器启动
通过脚本 start.sh 可以传递额外的参数,这些参数将被用于 podman run (例如:体积,变量,...)。
在容器中,工具被挂载到 /app/tools 并自动加载。
注
- I工具必须使用全局实例core_api和装饰器@core_api.mcp.tool。
- 不需要文件
__init__得益于动态加载器,.py 文件存在于工具子目录中。 - 所有工具加载过程中的日志和异常均通过LoggerHelper进行报告。
默认工具的贡献
要添加一个新工具:
- 在tools/目录下创建一个Python模块
- 定义一个用 @core_api.mcp.tool 装饰的函数
- 加载器将自动导入模块并注册该函数为可执行工具。
使用示例:
# tools/default/health_check.py
from mcp_core.core import core_api
@core_api.mcp.tool
def health_check() -> dict:
core_api.logger.info("Eseguo health_check")
return {"status": "ok"}
