汽车AI服务套件

Informant MCP服务器套件提供7个专用服务器,用于实时汽车数据和服务,包括诊断、导航、天气、维护、紧急情况、知识和车辆信息。

  • by informant-ai
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 informant-ai/informant-mcp-servers

使用说明

Informant MCP服务器-完整文档

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

目录

概述

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. 提交拉取请求

性能基准

响应时间

服务器 平均响应 第95个百分位数
诊断 12ms 25ms
导航 15ms 30ms
天气 8毫秒 18毫秒
维护 10ms 22ms
紧急 5ms 12ms
知识 3ms 8ms
车辆信息 7ms 15ms

内存使用

服务器 RAM使用率 CPU使用率
每台服务器 ~50MB ~5%
所有服务器 ~350MB ~35%

可扩展性

  • 并发请求:每台服务器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 <PID>
  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服务器:完整的汽车人工智能基础设施。