Mhlabs MCP Tools

🚀 mhlabs-mcp-tools

mhlabs-mcp-tools 是一个基于 FastMCP 构建的模块化 MCP 工具服务器。它提供了一个可扩展的 AI 工具生态系统，这些工具按功能类别（文本预处理、NLP 组件、文档分析等）组织，可以通过 MCP（模型上下文协议） 经由 STDIO 传输 进行动态加载和服务。

本项目是 MHLabs AI 代理生态系统 的一部分，旨在与 mhlabs-mcp-server、mhlabs-mcp-agents 以及下游 A2A 代理框架协同工作。

🚀 快速开始

开发环境搭建

1. 克隆并进入项目目录：

cd src/mhlabs_mcp_tools

2. 安装依赖：

pip install -r requirements.txt

3. 配置环境变量：

cp .env.example .env
# 编辑 .env 文件，填入你的配置信息

4. 启动服务器：

# 默认 STDIO 传输（适用于本地 MCP 客户端）
python mcp_server.py

# HTTP 传输（适用于基于 Web 的客户端）
python mcp_server.py --transport http --port 9000
# 或者，在安装 mhlabs-mcp-tools 后
python -m mhlabs_mcp_tools.mcp_server --transport http --port 9000

# 使用 FastMCP CLI（推荐）
fastmcp run mcp_server.py -t streamable-http --port 9000 -l DEBUG

# 禁用认证的调试模式
python mcp_server.py --transport http --debug --no-auth

传输选项

1. STDIO 传输（默认）

• 🔧 适用场景：本地工具、命令行集成、Claude Desktop
• 🚀 使用方法：python mcp_server.py 或 python mcp_server.py --transport stdio

2. HTTP（可流式传输）传输

• 🌐 适用场景：基于 Web 的部署、微服务、远程访问
• 🚀 使用方法：python mcp_server.py --transport http --port 9000
• 🌐 URL：http://127.0.0.1:9000/mcp/

3. SSE 传输（已弃用）

• ⚠️ 仅提供旧版支持，新项目建议使用 HTTP 传输
• 🚀 使用方法：python mcp_server.py --transport sse --port 9000

FastMCP CLI 使用方法

# 标准 HTTP 服务器
fastmcp run mcp_server.py -t streamable-http --port 9000 -l DEBUG

# 自定义主机
fastmcp run mcp_server.py -t streamable-http --host 0.0.0.0 --port 9000 -l DEBUG

# STDIO 传输（适用于本地客户端）
fastmcp run mcp_server.py -t stdio

# 使用 MCP 检查器的开发模式
fastmcp dev mcp_server.py -t streamable-http --port 9000

VS Code 开发

1. 在 VS Code 中打开项目：

code .

2. 使用调试配置：
   • Debug MCP Server (STDIO)：使用 STDIO 传输运行
   • Debug MCP Server (HTTP)：使用 HTTP 传输运行
   • Debug Tests：运行测试套件

✨ 主要特性

• FastMCP 服务器：纯 FastMCP 实现，支持多种传输协议
• 工厂模式：可复用的 MCP 工具工厂，便于服务管理
• 基于领域的组织方式：服务按业务领域（人力资源、技术支持等）组织
• 认证功能：可选的 Azure AD 认证支持
• 多种传输方式：支持 STDIO、HTTP（可流式传输）和 SSE 传输
• VS Code 集成：提供调试配置和开发设置
• 全面测试：使用 pytest 进行单元测试
• 灵活配置：基于环境的配置管理

📦 安装指南

环境变量

基于 .env.example 创建一个 .env 文件：

# 服务器设置
MCP_HOST=0.0.0.0
MCP_PORT=9000
MCP_DEBUG=false
MCP_SERVER_NAME=MHLABS MCP Server

# 认证设置
MCP_ENABLE_AUTH=true
AZURE_TENANT_ID=your-tenant-id-here
AZURE_CLIENT_ID=your-client-id-here
AZURE_JWKS_URI=https://login.microsoftonline.com/your-tenant-id/discovery/v2.0/keys
AZURE_ISSUER=https://sts.windows.net/your-tenant-id/
AZURE_AUDIENCE=api://your-client-id

认证

当 MCP_ENABLE_AUTH=true 时，服务器需要 Azure AD 承载令牌。请在 Azure 应用注册中进行相应的配置。 在开发环境中，可将 MCP_ENABLE_AUTH=false 设置为禁用认证。

💻 使用示例

基础用法

Python 客户端使用示例

import asyncio
from fastmcp import Client

client = Client("http://localhost:9000/mcp")

async def main():
    async with client:
        tools = await client.list_tools()
        # tools -> list[mcp.types.Tool]
        # print(tools)
        for tool in tools:
            print(f"Tool: {tool.name}")
        
        result = await client.call_tool("textprep.expand_contraction", {"input_text": "The must've SSN is 859-98-0987. The employee's phone number is 555-555-5555."})
        print("Result:", result)

asyncio.run(main())

命令行测试示例

# 测试服务器是否运行
curl http://localhost:9000/mcp/

# 使用 FastMCP CLI 进行测试
fastmcp dev mcp_server.py -t streamable-http --port 9000

高级用法

快速测试示例

测试 STDIO 传输：

# 以 STDIO 模式启动服务器
python mcp_server.py --debug --no-auth

# 使用 client_example.py 进行测试
python client_example.py

测试 HTTP 传输：

# 启动 HTTP 服务器
python mcp_server.py --transport http --port 9000 --debug --no-auth

# 使用 FastMCP 客户端进行测试
python -c "
from fastmcp import Client
import asyncio
async def test():
    async with Client('http://localhost:9000/mcp') as client:
        result = await client.call_tool(\"textprep.expand_contraction\", {\"input_text\": \"The must've SSN is 859-98-0987. The employee's phone number is 555-555-5555.\"})
        print(result)
asyncio.run(test())
"

使用 FastMCP CLI 进行测试：

# 使用 FastMCP CLI 启动服务器
fastmcp run mcp_server.py -t streamable-http --port 9000 -l DEBUG

# 服务器地址：http://127.0.0.1:9000/mcp/

📚 详细文档

可用服务

目前，该软件包主要分为三个模块：

1. NLP 组件

2. 文本预处理

该模块为用户提供了丰富的文本预处理工具：

添加新服务

1. 创建服务类：

from core.factory import MCPToolBase, Domain

class MyService(MCPToolBase):
    def __init__(self):
        super().__init__(Domain.MY_DOMAIN)

    def register_tools(self, mcp):
        @mcp.tool(tags={self.domain.value})
        async def my_tool(param: str) -> str:
            # 工具实现
            pass

    @property
    def tool_count(self) -> int:
        return 1  # 工具数量

2. 在服务器中注册：

# 在 mcp_server.py 中（会自动从 services/ 目录中注册）
factory.register_service(MyService())

3. 添加新领域（如果需要）：

# 在 core/factory.py 中
class Domain(Enum):
    # ... 现有领域
    MY_DOMAIN = "my_domain"

MCP 客户端使用

Python 客户端

import asyncio
from fastmcp import Client

client = Client("http://localhost:9000/mcp")

async def main():
    async with client:
        tools = await client.list_tools()
        # tools -> list[mcp.types.Tool]
        # print(tools)
        for tool in tools:
            print(f"Tool: {tool.name}")
        
        result = await client.call_tool("textprep.expand_contraction", {"input_text": "The must've SSN is 859-98-0987. The employee's phone number is 555-555-5555."})
        print("Result:", result)

asyncio.run(main())

命令行测试

# 测试服务器是否运行
curl http://localhost:9000/mcp/

# 使用 FastMCP CLI 进行测试
fastmcp dev mcp_server.py -t streamable-http --port 9000

🔧 技术细节

架构

mhlabs_mcp_tools/
├── .gitignore
├── .vscode/
│   └── settings.json
├── CHANGELOG.md
├── LICENSE
├── README.md
├── docs/
│   └── index.md
├── examples/
│   ├── example_client.py
│   └── example_client_http.py
├── mkdocs.yml
├── pyproject.toml
├── requirements.txt
├── server.json
└── src/
    ├── __init__.py
    ├── main.py
    └── mhlabs_mcp_tools/
        ├── __init__.py
        ├── core/
        │   ├── __init__.py
        │   ├── config.py
        │   ├── constants.py
        │   ├── factory.py
        │   └── prompts.py
        ├── data/
        │   ├── __init__.py
        │   ├── external/
        │   │   └── __init__.py
        │   ├── interim/
        │   │   └── __init__.py
        │   ├── processed/
        │   │   └── __init__.py
        │   └── raw/
        │       ├── __init__.py
        │       ├── contractions_dict.json
        │       ├── custom_substitutions.csv
        │       ├── leftovers_dict.json
        │       └── slang_dict.json
        ├── handlers/
        │   ├── __init__.py
        │   ├── custom_exceptions.py
        │   └── output_generator.py
        ├── mcp_server.py
        ├── models/
        │   └── __init__.py
        ├── nlp_components/
        │   ├── __init__.py
        │   └── nlp_model.py
        ├── services/
        │   ├── __init__.py
        │   ├── langchain_framework.py
        │   └── spacy_extractor.py
        └── text_preprocessing/
            ├── __init__.py
            ├── contractions.py
            ├── emo_unicode.py
            ├── slang_text.py
            └── text_preprocessing.py

服务器参数

usage: mcp_server.py [-h] [--transport {stdio,http,streamable-http,sse}]
                     [--host HOST] [--port PORT] [--debug] [--no-auth]

MHLABS MCP Server

options:
  -h, --help            show this help message and exit
  --transport, -t       Transport protocol (default: stdio)
  --host HOST           Host to bind to for HTTP transport (default: 127.0.0.1)
  --port, -p PORT       Port to bind to for HTTP transport (default: 9000)
  --debug               Enable debug mode
  --no-auth             Disable authentication

📄 许可证

本项目采用 MIT 许可证 © 2025 MusaddiqueHussain Labs

🤝 贡献指南

1. 遵循现有的代码结构和模式。
2. 为新功能添加测试。
3. 为新特性更新文档。
4. 使用提供的 VS Code 配置进行开发。

🧠 了解更多

• MCP 协议：https://modelcontextprotocol.io
• FastMCP GitHub：https://github.com/fastmcp/fastmcp
• LangGraph 集成指南（即将推出）

💡 使用建议

如果你想将 mhlabs-mcp-tools 嵌入到一个更大的基于 MCP 的编排器中，可以参考以下代码：

from fastmcp import StdioServerParameters
server_params = StdioServerParameters(
    command="python",
    args=["-m", "mhlabs_mcp_tools.server"],
    //env={"MHLABS_MCP_CATEGORY": "textprep,nlp"}
)

由 MusaddiqueHussain Labs 用心开发 ❤️