对等恢复MCP服务器
作为MCP（模型上下文协议）服务器交付的符合HIPAA标准（42 CFR第2部分）的生产级AI辅助对等恢复案例管理系统。
主要特点
- 42 CFR第2部分合规性：具有不可变审计日志的同意门控客户端数据访问
- 现场级AES-256加密：带零停机按键旋转的套圈
- 危机分类和升级：风险评分1-10，并通知主管
- SOAP注释生成：来自会话记录的人工智能临床文档
- 计费代码建议：按会话类型和持续时间匹配H码
- 资源匹配：识别障碍，将需求与可用服务相匹配
- 同伴角色扮演培训：基于场景的实践，有质量反馈
- 双重认证：基于TOTP的2FA，具有基于角色的访问控制
堆栈
- 框架：具有异步支持的FastMCP（mcp\[cli\]v1.26+）
- 数据库：PostgreSQL 16+SQLAlchemy 2.0异步+Alembic迁移
- 缓存：Redis 7用于JWT块列表、速率限制、会话管理
- 加密：密码学（Fernet AES-256-CBC）+python jose（JWT）
- 调度：pg_cron用于自动保留策略
- 语言：Python 3.11+
- 包管理器：紫外线
快速开始
先决条件
- Docker&Docker编写
- 或者：PostgreSQL 16、Redis 7、Python 3.11+
Docker（推荐）
# Set up environment
cp .env.example .env
# Edit .env with your JWT_SECRET_KEY and ENCRYPTION_KEY
# Start services
docker-compose up -d
# Apply migrations and start server
docker-compose logs app # Verify all healthy本地开发
# Install dependencies
uv sync
# Set up environment
cp .env.example .env
export $(grep -v '^#' .env | xargs)
# Start PostgreSQL + Redis (or use Docker)
# docker-compose up postgres redis
# Apply migrations
uv run alembic upgrade head
# Run server
uv run python -m peer_recovery.server项目结构
peer-recovery/
├── pyproject.toml # Dependencies, build config
├── .env.example # Environment variables
├── docker-compose.yml # Dev services
├── Dockerfile # App container
├── alembic/ # Database migrations
│ ├── env.py
│ ├── alembic.ini
│ └── versions/
│ ├── 001_initial_schema.py
│ ├── 002_audit_immutability.py
│ └── 003_pg_cron_retention.py
└── src/peer_recovery/
├── server.py # FastMCP entry point, lifespan, tool registration
├── config.py # Pydantic Settings
├── auth/
│ ├── tokens.py # JWT encode/decode
│ ├── totp.py # TOTP 2FA
│ ├── rbac.py # Role-based access control
│ ├── middleware.py # ASGI JWT + rate limiting
│ └── dependencies.py # require_auth(), require_role()
├── db/
│ ├── engine.py # create_async_engine, session factory
│ ├── models/ # ORM models
│ └── repositories/ # Async CRUD repositories
├── encryption/
│ ├── field_cipher.py # Fernet AES-256
│ ├── key_manager.py # Key rotation
│ └── encrypted_type.py # SQLAlchemy TypeDecorator
├── compliance/
│ ├── audit_logger.py # structlog + DB audit trail
│ └── consent.py # 42 CFR consent gates
├── ai/
│ ├── interfaces.py # Protocol ABCs + registry
│ └── mock/ # Keyword/template-based mock AI
├── tools/ # MCP tool implementations
└── services/ # Business logic (auth, clients, etc)MCP工具
所有工具都需要JWT身份验证。有些需要特定的角色或同意。
voice_to_notes
将音频转换为SOAP注释+计费代码。
- 需要：身份验证、客户端同意（访问）
- 退货：SOAP注释、计费代码、风险评分
crisis_triage
评估危机风险，必要时升级。
- 需要：身份验证
- 风险评分：1-10，如果>=7，则升级
- 退货：风险等级、行动、需要升级
client_monitor
跟踪客户进度并识别风险标志。
- 需要：身份验证、客户端同意（访问）
- 退货：趋势、风险标志、里程碑、恢复天数
resource_match
将客户需求与可用资源相匹配。
- 需要：身份验证、客户端同意（访问）
- 退货：匹配的资源、预测的障碍
ai_roleplay
使用AI反馈练习对等恢复场景。
- 需要：身份验证
- 退货：模拟响应、质量评分、反馈
export_notes
将会话笔记导出为PDF/DOCX/JSON。
- 需要：主管角色，客户出口同意
- 退货：文件数据，内容类型
身份验证流程
登录（初始）
POST /auth/login → {access_token, refresh_token, needs_2fa}访问令牌有效期15分钟，刷新令牌有效期7天。如果用户启用了TOTP：
- 访问令牌具有
2fa_verified: false - 仅
/auth/2fa/verify可使用此令牌访问
2FA验证
POST /auth/2fa/verify {user_id, totp_code} → {access_token, refresh_token}新代币有 2fa_verified: true.
令牌刷新
POST /auth/refresh {refresh_token} → {access_token}退出登录
POST /auth/logout → {}增加 jti 到Redis块列表，TTL=剩余令牌生命周期。
42 CFR第2部分合规性
同意要求
每个客户端数据访问都需要有效的 ConsentRecord:
consented_to=访问类型（访问、计费、研究、导出）valid_from≤现在≤valid_until（或valid_until为空）revoked_at为NULL
如果同意书缺失/无效→ ConsentError 已引发，操作记录为失败。
审计跟踪
每个操作都记录为不可变 audit_logs 表：
- 读取、创建、更新、删除、登录、导出、同意检查
- 包括参与者ID、资源类型/ID、客户端ID、IP、结果、详细JSON
- 数据库触发器阻止更新/删除
数据保留
pg_cron作业删除：
- 恢复期>90天
- 未升级的危机日志>90天
- 培训进度>1年
- 撤销同意书>2年
基于角色的访问控制
| 行动 | 同行 | 主管 | 管理员 |
|---|---|---|---|
| 阅读自己的客户 | ✓ | ✓ | ✓ |
| 阅读所有客户 | ✗ | ✓ | ✓ |
| 导出记录 | ✗ | ✓ | ✓ |
| 批准升级 | ✗ | ✓ | ✓ |
| 管理用户 | ✗ | ✗ | ✓ |
| 查看审核日志 | ✗ | ✗ | ✓ |
配置
通过环境变量（.env）进行的所有设置：
# Database
DATABASE_URL=postgresql+asyncpg://user:pass@host/db
DATABASE_ECHO=false
# Redis
REDIS_URL=redis://host:6379/0
# JWT
JWT_SECRET_KEY=
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=15
JWT_REFRESH_TOKEN_EXPIRE_DAYS=7
# Encryption
ENCRYPTION_KEY=
ENCRYPTION_KEY_OLD=
# AI Backend
AI_BACKEND=mock # mock, anthropic, huggingface
ANTHROPIC_API_KEY=
HUGGINGFACE_API_KEY=
# Server
MCP_SERVER_PORT=8000
DEBUG=false
LOG_LEVEL=info
# Feature Flags
ENABLE_2FA=true
ENABLE_AUDIT_LOGGING=true
ENABLE_FIELD_ENCRYPTION=true测试
# Run all tests
uv run pytest tests/ -v
# Coverage
uv run pytest tests/ --cov=src/peer_recovery --cov-report=html
# Integration tests (requires services running)
uv run pytest tests/integration/ -v迁移
# Apply all migrations
uv run alembic upgrade head
# Create new migration
uv run alembic revision --autogenerate -m "description"
# Rollback one revision
uv run alembic downgrade -1发展
生成加密密钥
from cryptography.fernet import Fernet
key = Fernet.generate_key()
print(key.decode()) # Set as ENCRYPTION_KEY in .env创建测试用户
uv run python -c "
from peer_recovery.services.auth_service import AuthService
from peer_recovery.config import get_settings
settings = get_settings()
# Use AuthService.register_user()
"安全说明
- 永远不要提交.env 用真钥匙
- JWT_SECRET_KEY 必须≥32个字符
- 加密密钥 必须是有效的base64 Fernet密钥
- 2FA 默认情况下为敏感操作启用
- 速率限制 每个用户或IP每分钟60个请求
- 审计日志 不可变；无法修改或删除
- PII加密 ORM层透明（加密文本）
合规
- ✓ HIPAA§164.312（a）（1）-访问控制（身份验证、2FA、RBAC）
- ✓ HIPAA§164.312（a）（2）-审计控制（Audit_logs，不可变）
- ✓ HIPAA§164.312（b）-审计日志记录（事件级日志记录）
- ✓ HIPAA§164.312（c）（1）-加密（静态AES-256，传输中的TLS）
- ✓ 42 CFR第2部分-同意要求（同意记录门）
- ✓ 42 CFR第2部分-数据保留（pg_cron删除策略）
支持
有关问题或疑问，请打开GitHub问题或联系恢复团队。
许可证
麻省理工学院