KYC MCP Server

KYC MCP服务器

用于KYC（了解您的客户）API的生产就绪模型上下文协议（MCP）服务器，集成了高级功能，包括自动工具注册、缓存、速率限制和全面的错误处理。

特性

• ✅ 自动工具注册表：从元数据JSON文件中自动发现工具
• ✅ 高级缓存：基于Redis的缓存，具有可配置的TTL
• ✅ 速率限制：使用令牌桶算法限制每个工具的速率
• ✅ JWT身份验证：具有令牌管理的安全API身份验证
• ✅ 重试逻辑：失败请求的指数回退
• ✅ 结构化日志记录：使用structlog进行全面记录
• ✅ Docker就绪：完全支持Docker和Docker Compose
• ✅ 类型安全：Pydantic v2用于数据验证
• ✅ 生产就绪：错误处理、监控和正常关机

已实施的工具

1.PAN验证(verify_pan)

验证PAN卡详细信息，姓名和出生日期是否匹配。

输入：

• pan：10个字符的PAN编号（例如“XXXPX1234A”）
• name_as_per_pan：PAN卡上的全名
• date_of_birth：出生日期，格式为DD/MM/YYYY
• consent：用户同意（“Y”或“Y”）
• reason：验证原因

输出：

• PAN验证状态
• 姓名和出生日期匹配结果
• Aadhaar播种状态
• PAN支架类别

缓存TTL： 1小时

2.PAN Aadhaar链路检查(check_pan_aadhaar_link)

检查PAN和Aadhaar是否链接。

输入：

• pan：单个PAN号码（第4个字符必须是“P”）
• aadhaar_number：12位Aadhaar数字
• consent：用户同意（“Y”或“Y”）
• reason：检查原因

输出：

• 链接状态（链接/未链接）
• 描述性信息

缓存TTL： 2小时

建筑

kyc-mcp-server/
├── src/
│   ├── main.py                 # Application entry point
│   ├── server/
│   │   └── mcp_server.py       # MCP server implementation
│   ├── tools/
│   │   ├── base_tool.py        # Abstract base tool class
│   │   ├── pan_verification.py # PAN verification tool
│   │   └── pan_aadhaar_link.py # PAN-Aadhaar link tool
│   ├── registry/
│   │   └── tool_registry.py    # Auto tool discovery & registration
│   ├── clients/
│   │   └── kyc_api_client.py   # KYC API client with retry logic
│   ├── auth/
│   │   └── jwt_manager.py      # JWT token management
│   ├── cache/
│   │   └── redis_cache.py      # Redis caching layer
│   ├── models/
│   │   ├── requests.py         # Request models
│   │   └── responses.py        # Response models
│   └── utils/
│       ├── logger.py           # Structured logging
│       └── rate_limiter.py     # Rate limiting
├── config/
│   └── settings.py             # Configuration management
├── metadata/
│   └── tools/                  # Tool metadata JSON files
│       ├── pan_verification.json
│       └── pan_aadhaar_link.json
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env.example

安装

先决条件

• Python 3.11+
• Redis（用于缓存）
• KYC API证书

本地设置

1. 克隆存储库

git clone 
cd kyc-mcp-server

1. 创建虚拟环境

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

1. 安装依赖项

pip install -r requirements.txt

1. 配置环境

cp .env.example .env
# Edit .env with your credentials

1. 启动Redis

docker run -d -p 6379:6379 --name kyc-redis redis:7-alpine

1. 运行服务器

python -m src.main

Docker设置

1. 配置环境

cp .env.example .env
# Edit .env with your credentials

1. 使用Docker Compose构建和运行

docker-compose up -d

1. 查看日志

docker-compose logs -f kyc-mcp-server

1. 停止服务器

docker-compose down

配置

所有配置都通过环境变量进行管理。看 .env.example 所有可用选项。

关键配置选项

用法

使用MCP客户端

# Connect to the server
mcp-client connect stdio -- python -m src.main

# List available tools
mcp-client list-tools

# Call a tool
mcp-client call-tool verify_pan '{
  "pan": "XXXPX1234A",
  "name_as_per_pan": "John Doe",
  "date_of_birth": "01/01/1990",
  "consent": "Y",
  "reason": "KYC verification"
}'

工具调用示例

PAN验证

{
  "tool": "verify_pan",
  "arguments": {
    "pan": "XXXPX1234A",
    "name_as_per_pan": "John Doe",
    "date_of_birth": "01/01/1990",
    "consent": "Y",
    "reason": "Customer onboarding"
  }
}

答复：

{
  "pan": "XXXPX1234A",
  "category": "individual",
  "status": "valid",
  "remarks": null,
  "name_match": true,
  "dob_match": true,
  "aadhaar_seeding_status": "y",
  "verified_at": 1234567890,
  "_cached": false
}

PAN Aadhaar链路检查

{
  "tool": "check_pan_aadhaar_link",
  "arguments": {
    "pan": "XXXPX1234A",
    "aadhaar_number": "123456789012",
    "consent": "Y",
    "reason": "Link verification"
  }
}

答复：

{
  "linked": true,
  "status": "y",
  "message": "PAN and Aadhaar are linked",
  "checked_at": 1234567890,
  "_cached": false
}

添加新工具

服务器使用自动工具注册表系统。要添加新工具，请执行以下操作：

1. 创建工具类 在 src/tools/

from src.tools.base_tool import BaseTool

class NewTool(BaseTool):
    def get_name(self) -> str:
        return "new_tool_name"
    
    async def execute(self, params):
        # Implementation
        pass

1. 创建元数据文件 在 metadata/tools/

{
  "name": "new_tool_name",
  "description": "Tool description",
  "input_schema": { ... },
  "output_schema": { ... }
}

1. 注册工具 在 src/main.py

new_tool = NewTool(api_client=self.api_client)
tool_registry.register_tool(new_tool)

1. 重新启动服务器 -工具自动可用！

错误处理

服务器提供全面的错误处理：

• 验证错误：输入参数无效
• 超出速率限制：超出费率限制
• TOOL_NOT_FOUND：请求的工具未知
• 执行错误：工具执行失败
• 服务可用：外部API不可用

所有错误都包括描述性消息和相应的错误代码。

监控

日志

结构化JSON日志输出到stdout：

{
  "event": "tool_executed_successfully",
  "tool": "verify_pan",
  "timestamp": "2024-01-20T10:30:00Z",
  "level": "info"
}

指标

服务器在端口9090上公开与Prometheus兼容的指标（可配置）。

演出

• 缓存命中率：重复查询率>70%
• 响应时间：对于未缓存的请求，\<500ms（p95）
• 响应时间：对于缓存的请求，\<10ms（p95）
• 并发请求：支持100多个并发请求

安全

• API调用基于JWT的身份验证
• 使用Pydantic进行输入验证
• 限制利率以防止滥用
• 通过环境变量实现安全的凭据管理
• 日志中没有敏感数据

故障排除

Redis连接失败

# Check if Redis is running
docker ps | grep redis

# Start Redis
docker run -d -p 6379:6379 redis:7-alpine

超出费率限制

# Increase rate limits in .env
RATE_LIMIT_PER_MINUTE=120
RATE_LIMIT_PER_HOUR=2000

未找到工具

# Check metadata files exist
ls metadata/tools/

# Check tool registration in logs
docker-compose logs kyc-mcp-server | grep "tool_registered"

发展

运行测试

# Install dev dependencies
pip install -r requirements-dev.txt

# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=src --cov-report=html

代码质量

# Format code
black src/

# Lint code
ruff check src/

# Type checking
mypy src/

许可证

\[在此处添加您的许可证\]

支持

对于问题和疑问：

• 在存储库中创建问题
• 联系人：\[your-email@example.com\]

致谢

内置：

• MCP-SDK
• 派丹蒂克
• HTTPX
• 瑞迪斯
• 结构日志