AI导哈

多协议AI工具服务器

一个支持REST API、MCP协议和WebSocket的多协议AI工具服务器，提供简单装饰器、类型安全和生产级功能。

  • by AliQambari
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 AliQambari/unified-mcp-server

使用说明

🚀 统一MCP服务器 - 一个工具服务器；支持多种协议

一个简单的服务器，能够无缝地提供AI工具和资源 多个协议REST API、MCP（模型上下文协议）和WebSocket连接。

✨ 特点/功能

🔌（电源插头/插座） 三重协议支持

  • **REST API（Representational State Transfer Application Programming Interface，表述性状态传递应用程序编程接口）**用于网页集成的标准HTTP端点
  • **通过HTTP的MCP（可能指某种协议或系统，具体需根据上下文确定，此处直译为“MCP over HTTP”）**人工智能助手（如Claude等）的模型上下文协议
  • WebSocket实时双向通信

🎯（目标/瞄准） 开发者体验

  • 简单装饰器： @tool， @resource， @resource_template， @prompt - 就这样！
  • 类型安全带有mypy支持的完整类型提示
  • **异步/等待（Async/Await）**全程原生支持异步功能

🏗️（建筑工地，表示建筑或施工的场景） 已准备好投入生产

  • 全面日志记录具有可配置级别的结构化日志记录
  • 错误处理优雅的错误响应与恢复
  • CORS 支持处理了跨域请求
  • 健康检查内置监控端点

📦 安装

使用紫外线（推荐）

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create and activate virtual environment
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install the package
uv add unified-mcp-server

使用 pip

pip install unified-mcp-server

开发安装

# Clone the repository
git clone <repository-url>
cd unified-mcp-server

# Using uv (recommended)
uv sync --dev

# Or using pip
pip install -e ".[dev]"

🚀 快速入门

基本示例

from unified_server import create_server, tool, resource, prompt

# Define tools with simple decorators
@tool(description="Add two numbers together")
def add(a: int, b: int) -> int:
    """Add two integers and return the result"""
    return a + b

@tool(description="Analyze text sentiment")
def analyze_sentiment(text: str) -> dict:
    """Analyze the sentiment of given text"""
    # Your sentiment analysis logic here
    return {"sentiment": "positive", "confidence": 0.95}

# Define resources (data sources)
@resource(
    uri="config://app/settings",
    description="Application configuration",
    mime_type="application/json"
)
def get_config():
    return {
        "app_name": "My App",
        "version": "1.0.0",
        "features": {"ai_enabled": True}
    }

# Define prompts for AI interactions
@prompt(description="Code review prompt")
def code_review_prompt(language: str):
    return [{
        "role": "user",
        "content": {
            "type": "text",
            "text": f"Review this {language} code for best practices"
        }
    }]

# Create and run server
if __name__ == "__main__":
    server = create_server(name="my-server", version="1.0.0")
    server.run(host="0.0.0.0", port=8000)

📁 完整示例

见 src/tool_server.py 以下是一个包含全面示例的：

  • 多种工具（数学工具、搜索工具、情感分析工具）
  • 各种资源（配置、用户数据、文档）
  • 带有参数的高级提示
  • 实际文件加载
  • 错误处理

🔧 使用示例

🌐 REST API

# List all available tools
curl http://localhost:8000/tools

# Execute a tool
curl -X POST http://localhost:8000/tools/add \
  -H "Content-Type: application/json" \
  -d '{"a": 15, "b": 27}'

# Get all resources
curl http://localhost:8000/resources

# Read a specific resource
curl http://localhost:8000/resources/get_config

# List available prompts
curl http://localhost:8000/prompts

# Generate a prompt
curl -X POST http://localhost:8000/prompts/code_review_prompt \
  -H "Content-Type: application/json" \
  -d '{"language": "python"}'

🤖 MCP（微型计算机处理器/模块化控制系统等，具体含义根据上下文确定）集成

Claude 桌面配置

{
  "mcpServers": {
    "unified-server": {
      "command": "npx",
      "args": ["@modelcontextprotocol/server-everything", "http://localhost:8000/mcp"]
    }
  }
}

直接MCP客户端

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Connect to your unified server
    async with stdio_client(StdioServerParameters(
        command="python",
        args=["-m", "your_server_module"]
    )) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            
            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[tool.name for tool in tools.tools]}")
            
            # Call a tool
            result = await session.call_tool("add", {"a": 10, "b": 20})
            print(f"Result: {result.content}")

asyncio.run(main())

🔌 WebSocket 连接

// Connect via WebSocket for real-time communication
const ws = new WebSocket('ws://localhost:8000/ws');

ws.onopen = function() {
    // Send tool execution request
    ws.send(JSON.stringify({
        type: 'tool_call',
        tool: 'add',
        parameters: { a: 5, b: 3 }
    }));
};

ws.onmessage = function(event) {
    const response = JSON.parse(event.data);
    console.log('Tool result:', response.result);
};

📚 示例

基本用法

生产示例

  • src/tool_server.py - 功能齐全的服务器，具备：
    • 🔧（扳手或维修工具的符号，无具体文字含义） 工具数学运算、搜索、情感分析
    • 📄（文件/纸张） 资源配置、用户数据、文档
    • 💬（表示对话或聊天的符号，无具体含义，可理解为“对话”或“聊天”） 提示系统提示、代码审查、调试
    • 📁 文件夹 文件操作从磁盘加载真实文件

🏗️ 建筑学

src/unified_server/
├── 🏛️ core/              # Core server and registry
│   ├── server.py         # Main FastAPI server
│   ├── registry.py       # Tool/resource registry
│   └── config.py         # Configuration management
├── 🎨 decorators/         # Decorator implementations
│   ├── tool.py          # @tool decorator
│   ├── resource.py      # @resource decorator
│   ├── resource_template.py  # @resource template decorator
│   └── prompt.py        # @prompt decorator
├── 🛣️ routes/            # HTTP route handlers
│   ├── tools.py         # Tool execution endpoints
│   ├── resources.py     # Resource access endpoints
│   ├── prompts.py       # Prompt generation endpoints
│   └── mcp.py           # MCP protocol endpoints
├── 🔧 handlers/          # Protocol handlers
│   └── mcp_handlers.py  # MCP message handling
└── 🛠️ utils/             # Utilities
    ├── inspection.py    # Function signature analysis
    └── logging.py       # Logging configuration

根据上面的信息，执行如下指令：

一旦您的服务器运行起来，请访问：

  • 📖 交互式文档： http://localhost:8000/docs (Swagger UI)
  • 📋 ReDoc（可译为“重新记录”或根据上下文具体含义调整，此处保留原英文形式以体现专业术语）： http://localhost:8000/redoc （替代文档）
  • 🔧 OpenAPI 规格（或模式）： http://localhost:8000/openapi.json

🛠️ 开发

设置开发环境

# Using uv (recommended)
uv sync --dev
source .venv/bin/activate

# Using pip
pip install -e ".[dev]"

运行测试

# Run all tests
pytest

# Run with coverage
pytest --cov=unified_server

# Run specific test file
pytest tests/test_tools.py -v

代码质量

# Format code
black src tests examples

# Lint code
ruff check src tests examples

# Type checking
mypy src

# Run all quality checks
make lint  # if using the provided Makefile

项目命令

# Start development server with auto-reload
uv run python src/tool_server.py

# Run basic example
uv run python examples/basic_example.py

# Run advanced example
uv run python examples/advanced_example.py

🐳 Docker 支持

# Build image
docker build -t unified-mcp-server .

# Run container
docker run -p 8000:8000 unified-mcp-server

# Using docker-compose
docker-compose up

🤝 贡献（或：参与贡献）

  1. 叉子 存储库
  2. 创建 特性分支： git checkout -b feature/amazing-feature
  3. 提交（或“承诺”） 您的更改： git commit -m 'Add amazing feature'
  4. 推动；推送 到分公司： git push origin feature/amazing-feature
  5. 打开 一个拉取请求

📄 许可证

这个项目遵循以下许可协议 麻省理工学院许可证 - 看见 许可证 详情见文件。

🙏 致谢

  • FastAPI 对于出色的网络框架
  • MCP协议 用于标准化人工智能工具接口
  • Pydantic（注：这是一个编程库的名称，直接翻译为中文时通常保留原名，表示一种用于数据验证和设置的Python库） 用于数据验证和序列化
  • Uvicorn 用于ASGI服务器实现

相关 MCP