Informant MCP Servers

Informant MCP服务器-完整文档

   

用于汽车人工智能的完整MCP服务器套件 -7台提供实时汽车数据和服务的专用服务器。

目录

• 概述
• 建筑
• MCP服务器
• 快速开始
• 安装
• API文档
• 测试
• 生产部署
• 发展

概述

Informant MCP First架构使用 7台专用MCP服务器 而不是单一的大型语言模型。这种方法提供了：

• 快100倍 响应（15ms vs 1500ms）
• 100%准确 数据（无幻觉）
• 边缘就绪 部署（1MB意图路由器）
• 始终为最新 信息（实时API）
• 模块化设计 （独立更新服务器）

建筑

graph TD
    A[User Query] --> B[Intent Router 1MB]
    B --> C{Intent Classification}
    C -->|diagnostic_*| D[Diagnostic Server :8001]
    C -->|navigation_*| E[Navigation Server :8002]
    C -->|weather_*| F[Weather Server :8003]
    C -->|maintenance_*| G[Maintenance Server :8004]
    C -->|emergency_*| H[Emergency Server :8005]
    C -->|knowledge_*| I[Knowledge Server :8006]
    C -->|vehicle_*| J[Vehicle Info Server :8007]
    
    D --> K[Real Response]
    E --> K
    F --> K
    G --> K
    H --> K
    I --> K
    J --> K

MCP服务器

1.诊断MCP服务器 -端口8001

目的：车辆诊断、OBD-II分析、故障排除

特性:

• ✅ OBD-II错误代码数据库（P0171、P0300等）
• ✅ 发动机症状分析
• ✅ 制动系统诊断
• ✅ 轮胎状况评估
• ✅ 维修成本估算

示例用法:

POST http://localhost:8001/mcp
{
  "intent": "diagnostic_engine",
  "query": "My check engine light is on",
  "context": {}
}

响应:

{
  "success": true,
  "data": {
    "diagnosis": {
      "likely_code": "P0171",
      "description": "System Too Lean (Bank 1)",
      "severity": "medium",
      "estimated_cost": "$50-200"
    }
  }
}

2.导航MCP服务器 -端口8002

目的：位置服务、路线、交通信息

特性:

• ✅ 查找附近的位置（加油站、机械师、医院）
• ✅ 考虑交通因素的路线规划
• ✅ ETA计算
• ✅ 避开收费公路
• ✅ 紧急定位

示例用法:

POST http://localhost:8002/mcp
{
  "intent": "navigation_location",
  "query": "Find nearest gas station",
  "context": {}
}

3.气象MCP服务器 -端口8003

目的：影响驾驶安全的天气条件

特性:

• ✅ 按天气进行驾驶安全分析
• ✅ 特定路线的天气预报
• ✅ 风险评估（1-10分）
• ✅ 旅行建议
• ✅ 能见度和路况报告

示例用法:

POST http://localhost:8003/mcp
{
  "intent": "weather_driving",
  "query": "Is it safe to drive in this weather?",
  "context": {}
}

4.维护MCP服务器 -端口8004

目的：车辆维护计划和服务历史

特性:

• ✅ 保养间隔计算
• ✅ 维护成本估算
• ✅ 车辆特定时间表
• ✅ 服务历史跟踪
• ✅ 维修预算分析

示例用法:

POST http://localhost:8004/mcp
{
  "intent": "maintenance_schedule",
  "query": "When is my next oil change due?",
  "context": {"vehicle": {"mileage": 45000}}
}

5.应急MCP服务器 -端口8005

目的：应急响应和道路救援

特性:

• ✅ 应急协议生成
• ✅ 道路救援协调
• ✅ 按场景划分的安全说明
• ✅ 服务提供商建议
• ✅ 特定位置的安全考虑

示例用法:

POST http://localhost:8005/mcp
{
  "intent": "emergency_roadside",
  "query": "Car broke down on highway",
  "context": {"location": "highway"}
}

6.知识MCP服务器 -端口8006

目的：汽车操作指南和教育内容

特性:

• ✅ 分步维修指南
• ✅ 警告灯说明
• ✅ 维护提示和建议
• ✅ 工具要求
• ✅ 安全警告

示例用法:

POST http://localhost:8006/mcp
{
  "intent": "knowledge_general",
  "query": "How to change a tire",
  "context": {}
}

7.车辆信息MCP服务器 -端口8007

目的：车辆数据、规格和登记信息

特性:

• ✅ 车辆规格查询
• ✅ VIN解码
• ✅ 注册信息
• ✅ 召回通知
• ✅ 维修历史
• ✅ 车辆估价

示例用法:

POST http://localhost:8007/mcp
{
  "intent": "vehicle_info",
  "query": "What are my car specifications?",
  "context": {}
}

快速开始

1.克隆和安装

git clone https://github.com/informant-ai/informant-mcp-servers
cd informant-mcp-servers
pip install -r requirements.txt

2.启动所有MCP服务器

python start_mcp_servers.py

预期产量：

🚀 Starting Informant MCP-First Architecture
==================================================
🔧 Starting Diagnostic MCP Server on port 8001...
✅ Diagnostic MCP Server started successfully
🗺️ Starting Navigation MCP Server on port 8002...
✅ Navigation MCP Server started successfully
🌤️ Starting Weather MCP Server on port 8003...
✅ Weather MCP Server started successfully
🔧 Starting Maintenance MCP Server on port 8004...
✅ Maintenance MCP Server started successfully
🚨 Starting Emergency MCP Server on port 8005...
✅ Emergency MCP Server started successfully
📚 Starting Knowledge MCP Server on port 8006...
✅ Knowledge MCP Server started successfully
🚗 Starting Vehicle Info MCP Server on port 8007...
✅ Vehicle Info MCP Server started successfully

🎉 All MCP servers started successfully!

3.测试系统

python test_mcp_system.py

4.启动意图路由器

python inference.py

安装

系统要求

• Python 3.9或更高版本
• 最低2GB RAM（建议4GB）
• 1GB磁盘空间
• 外部API的互联网连接

依赖关系安装

基本安装:

pip install fastapi uvicorn pydantic aiohttp torch transformers

完整安装:

pip install -r requirements.txt

开发安装:

pip install -r requirements.txt
pip install black isort mypy pytest-cov pre-commit

验证

python -c "import fastapi, uvicorn, torch, transformers; print('✅ All dependencies installed')"

API文档

通用MCP请求格式

{
  "intent": "string",      # Intent classification
  "query": "string",       # User's natural language query
  "context": {             # Optional context information
    "location": "string",
    "vehicle": {...},
    "user_id": "string"
  }
}

通用MCP响应格式

{
  "success": boolean,      # Request success status
  "data": {               # Response data (if successful)
    "type": "string",
    "timestamp": "ISO8601",
    ...
  },
  "error": "string"       # Error message (if failed)
}

健康检查端点

所有服务器都提供健康检查端点：

GET http://localhost:800X/health

答复：

{
  "status": "healthy",
  "service": "diagnostic_mcp_server"
}

交互式API文档

每台服务器都提供交互式Swagger文档：

• 诊断的：http://localhost:8001/docs
• 导航：http://localhost:8002/docs
• 天气：http://localhost:8003/docs
• 维护：http://localhost:8004/docs
• 紧急情况：http://localhost:8005/docs
• 知识：http://localhost:8006/docs
• 车辆信息：http://localhost:8007/docs

测试

自动化测试

# Test all servers
python test_mcp_system.py

# Test individual servers
curl -X POST http://localhost:8001/mcp \
  -H "Content-Type: application/json" \
  -d '{"intent":"diagnostic_engine","query":"check engine light","context":{}}'

手动测试示例

诊断服务器:

curl -X POST http://localhost:8001/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "intent": "diagnostic_brake",
    "query": "brakes making grinding noise",
    "context": {}
  }'

导航服务器:

curl -X POST http://localhost:8002/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "intent": "navigation_location",
    "query": "find nearest mechanic",
    "context": {}
  }'

天气服务器:

curl -X POST http://localhost:8003/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "intent": "weather_driving",
    "query": "weather conditions for road trip",
    "context": {}
  }'

负载测试

# Install siege for load testing
pip install locust

# Run load test
locust -f load_test.py --host=http://localhost:8001

生产部署

Docker部署

# docker-compose.yml
version: '3.8'
services:
  diagnostic-server:
    build: .
    command: python diagnostic_mcp_server.py
    ports:
      - "8001:8001"
    
  navigation-server:
    build: .
    command: python navigation_mcp_server.py
    ports:
      - "8002:8002"
  
  # ... other servers

docker-compose up -d

Kubernetes部署

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-servers
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-servers
  template:
    metadata:
      labels:
        app: mcp-servers
    spec:
      containers:
      - name: diagnostic-server
        image: informant/diagnostic-server:latest
        ports:
        - containerPort: 8001

环境变量

# .env file
MCP_LOG_LEVEL=INFO
MCP_HOST=0.0.0.0
MCP_WORKERS=4
MCP_TIMEOUT=30

# External API keys (optional)
GOOGLE_MAPS_API_KEY=your_key_here
OPENWEATHER_API_KEY=your_key_here
OBD_API_KEY=your_key_here

生产配置

# production.py
import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        "diagnostic_mcp_server:app",
        host="0.0.0.0",
        port=8001,
        workers=4,
        access_log=True,
        log_level="info"
    )

发展

项目结构

informant-mcp-servers/
├── diagnostic_mcp_server.py      # Diagnostic MCP server
├── navigation_mcp_server.py      # Navigation MCP server  
├── weather_mcp_server.py         # Weather MCP server
├── maintenance_mcp_server.py     # Maintenance MCP server
├── emergency_mcp_server.py       # Emergency MCP server
├── knowledge_mcp_server.py       # Knowledge MCP server
├── vehicle_mcp_server.py         # Vehicle info MCP server
├── start_all_mcp_servers.py      # Server management script
├── test_mcp_system.py           # Testing script
├── requirements.txt              # Dependencies
├── README.md                     # This documentation
├── .env.example                  # Environment variables template
├── docker-compose.yml            # Docker deployment
└── tests/                        # Test files
    ├── test_diagnostic.py
    ├── test_navigation.py
    └── ...

添加新的MCP服务器

1. 创建新的服务器文件： new_mcp_server.py
2. 遵循标准MCP格式：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="New MCP Server")

class MCPRequest(BaseModel):
    intent: str
    query: str
    context: Optional[Dict] = {}

@app.post("/mcp")
async def handle_request(request: MCPRequest):
    # Your logic here
    return {"success": True, "data": {...}}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8008)

1. 增添 start_all_mcp_servers.py
2. 更新测试

代码的风格

# Format code
black *.py
isort *.py

# Type checking
mypy *.py

# Linting
flake8 *.py

贡献

1. 分叉存储库
2. 创建要素分支： git checkout -b feature/new-server
3. 进行更改并添加测试
4. 运行测试套件： python test_mcp_system.py
5. 提交拉取请求

性能基准

响应时间

内存使用

可扩展性

• 并发请求：每台服务器1000+
• 水平缩放：添加更多服务器实例
• 负载平衡：使用nginx或云负载均衡器

安全

认证 （可选）

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer

security = HTTPBearer()

@app.post("/mcp")
async def protected_endpoint(request: MCPRequest, token: str = Depends(security)):
    # Validate token
    if not validate_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    # ... rest of logic

速率限制

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/mcp")
@limiter.limit("100/minute")
async def rate_limited_endpoint(request: Request, mcp_request: MCPRequest):
    # ... logic

支持

故障排除

常见问题:

1. 端口已在使用中:

# Find process using port
lsof -i :8001
# Kill process
kill -9 

1. 导入错误:

# Reinstall dependencies
pip install --force-reinstall -r requirements.txt

1. 服务器无法启动:

# Check logs
python diagnostic_mcp_server.py
# Check port availability
netstat -tulpn | grep 8001

获取帮助

• 📋 问题:
• 💬 讨论:
• 📖 文档: 完整文档

许可证

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

致谢

• 模型上下文协议：革命性的MCP架构
• 快速 API：优秀的web框架
• 汽车社区：用于领域专业知识和验证
• 贡献者：所有帮助改进这个项目的人

______________________________________________________________________

专为汽车人工智能的未来而打造——专业、快速、准确且始终保持最新。

*Informant MCP服务器：完整的汽车人工智能基础设施。*