MCP防护装置
模型上下文协议服务器的工具级信任实施。
](https://badge.fury.io/py/capiscio-mcp)  
MCP防护装置 (pip install capiscio-mcp)提供信任徽章和身份验证 模型上下文协议(MCP) 工具调用。它实现了:
- RFC-006:MCP工具权威和证据
- RFC-007:MCP服务器身份披露和验证
安装
pip install capiscio-mcp对于MCP SDK集成(FastMCP包装器):
pip install capiscio-mcp[mcp]为什么选择MCP Guard?
MCP服务器向自主代理提供强大的工具——文件系统、数据库、API。但MCP本身并没有定义如何:
- 验证 哪个代理正在调用工具
- 授权 该代理是否应该有访问权限
- 审计 事件后审查发生了什么
MCP Guard通过以下方式解决了这个问题:
| 特性 | 描述 |
|---|---|
| @警卫装饰师 | 保护具有信任级别要求的工具 |
| 证据记录 | 每次调用的加密审计跟踪 |
| 服务器标识 | 连接前验证MCP服务器 |
| 服务器注册 | 生成密钥对并注册服务器DID |
| 信任级别 | 0(自签名)→ 4 (扩展验证) |
快速入门
构建MCP服务器? 从开始 快速入门1. 连接到MCP服务器? 从开始 快速入门2. 注册服务器身份? 从开始 快速入门3.
快速入门1:服务器端(工具保护)
使用信任级别要求保护您的MCP工具:
from capiscio_mcp import guard
@guard(min_trust_level=2)
async def read_database(query: str) -> list[dict]:
"""Only agents with Trust Level 2+ can execute this tool."""
# ... database query logic
pass
# Sync version available
from capiscio_mcp import guard_sync
@guard_sync(min_trust_level=2)
def read_database_sync(query: str) -> list[dict]:
pass具有完整配置
from capiscio_mcp import guard, GuardConfig
config = GuardConfig(
min_trust_level=2,
trusted_issuers=["did:web:registry.capisc.io"],
allowed_tools=["read_*", "list_*"],
require_badge=True, # Deny anonymous access
)
@guard(config=config)
async def execute_query(sql: str) -> list[dict]:
pass快速入门2:客户端(服务器验证)
验证您连接到的MCP服务器的身份:
from capiscio_mcp import verify_server, ServerState
result = await verify_server(
server_did="did:web:mcp.example.com",
server_badge="eyJhbGc...",
transport_origin="https://mcp.example.com",
)
if result.state == ServerState.VERIFIED_PRINCIPAL:
print(f"Trusted server at Level {result.trust_level}")
elif result.state == ServerState.DECLARED_PRINCIPAL:
print("Server identity declared but not verified")
elif result.state == ServerState.UNVERIFIED_ORIGIN:
print("Warning: Server did not disclose identity")快速入门3:服务器注册
在CapiscIO注册表中注册MCP服务器的身份:
from capiscio_mcp import setup_server_identity
# One-step setup: generate keys + register with registry
result = await setup_server_identity(
server_id="550e8400-e29b-41d4-a716-446655440000", # From dashboard
api_key="sk_live_...", # Registry API key
ca_url="https://registry.capisc.io", # Optional, defaults to production
output_dir="./keys",
)
print(f"Server DID: {result['did']}")
# did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK
print(f"Private key saved to: {result['private_key_path']}")分步注册
from capiscio_mcp import generate_server_keypair, register_server_identity
# Step 1: Generate keypair
keys = await generate_server_keypair(output_dir="./keys")
# Step 2: Register with registry
await register_server_identity(
server_id="550e8400-e29b-41d4-a716-446655440000",
api_key="sk_live_...",
did=keys["did_key"],
public_key=keys["public_key_pem"],
ca_url="https://registry.capisc.io", # Optional, defaults to production
)MCP SDK集成
与官方无缝集成 MCP Python SDK,与一起安装 mcp 额外:
pip install capiscio-mcp[mcp]配备FastMCP包装器的服务器
创建具有内置信任强制的MCP服务器:
from capiscio_mcp.integrations.mcp import CapiscioMCPServer
# db is your application's database connection (asyncpg, databases, etc.)
db = ... # e.g. databases.Database("postgresql://...")
server = CapiscioMCPServer.connect()
@server.tool(min_trust_level=2)
async def get_user(user_id: int) -> dict:
"""Only agents with Trust Level 2+ can read user data."""
return await db.fetch_one("SELECT * FROM users WHERE id = $1", user_id)
@server.tool(min_trust_level=1)
async def list_tables() -> list[str]:
"""Agents with a valid badge (Trust Level 1+) can list tables."""
return await db.get_table_names()
# Run the server (stdio transport)
server.run()具有信任验证的客户端
通过stdio传输连接到MCP服务器:
from capiscio_mcp.integrations.mcp import CapiscioMCPClient
async with CapiscioMCPClient(
command="python",
args=["my_mcp_server.py"],
min_trust_level=1,
badge="eyJhbGc...", # Your client badge
) as client:
# List available tools
tools = await client.list_tools()
print(f"Available tools: {[t['name'] for t in tools]}")
# Call a tool
result = await client.call_tool("read_file", {"path": "/data/config.json"})
print(result)CapiscioMCPServer.connect()--“让我们加密”样式设置
注册您的MCP服务器,只需一次呼叫即可获得徽章:
from capiscio_mcp.integrations.mcp import CapiscioMCPServer
server = CapiscioMCPServer.connect()
print(server.did) # did:web:registry.capisc.io:servers:550e8400-...
print(server.badge) # Current badge JWS (auto-issued)使用环境变量
server = CapiscioMCPServer.connect()| 变量 | 必填 | 描述 |
|---|---|---|
CAPISCIO_SERVER_ID | 是 | 来自仪表板的服务器UUID |
CAPISCIO_API_KEY | 是 | 注册表API键 |
CAPISCIO_SERVER_URL | 否 | 注册表URL(默认:生产) |
CAPISCIO_SERVER_DOMAIN | 否 | 颁发徽章的域 |
CAPISCIO_SERVER_PRIVATE_KEY_PEM | 否 | 适用于临时环境的PEM编码Ed25519私钥 |
部署到容器/无服务器
在临时环境(Docker、Lambda、Cloud Run)中,本地 ~/.capiscio/ 目录 重启后无法存活。首次运行时,SDK会生成一个密钥对并记录一个捕获提示:
╔══════════════════════════════════════════════════════════╗
║ New server identity generated — save key for persistence ║
╚══════════════════════════════════════════════════════════╝
Add to your secrets manager / .env:
CAPISCIO_SERVER_PRIVATE_KEY_PEM='-----BEGIN PRIVATE KEY-----\nMC4C...\n-----END PRIVATE KEY-----\n'将该值复制到机密管理器中,并将其设置为环境变量。 在后续启动时,SDK将恢复相同的DID,而不会生成新的标识。
关键分辨率优先级: 环境变量→ 本地文件→ 生成新的。
# docker-compose.yml
services:
mcp-server:
environment:
CAPISCIO_SERVER_ID: "550e8400-..."
CAPISCIO_API_KEY: "sk_live_..."
CAPISCIO_SERVER_PRIVATE_KEY_PEM: "${MCP_SERVER_KEY}" # from secrets看 部署指导 如需查看完整示例。
核心连接模式
MCP Guard连接到capiscio核心进行加密操作:
嵌入式模式(默认)
SDK自动下载并管理核心二进制文件:
pip install capiscio-mcp
# Just works! Binary downloaded on first use.外部模式
连接到单独管理的核心服务:
# Start core in another terminal
capiscio mcp serve --listen localhost:50051
# SDK connects to external core
export CAPISCIO_CORE_ADDR="localhost:50051"信任级别
根据RFC-002 v1.4:
| 级别 | 名称 | 验证 | 用例 |
|---|---|---|---|
| 0 | 自签名(SS) | 无, did:key 发行人 | 本地开发、测试、演示 |
| 1 | 注册(REG) | 账户注册 | 开发、内部代理 |
| 2 | 域名验证(DV) | DNS/HTTP挑战 | 生产、B2B代理 |
| 3 | 组织验证(OV) | DUNS/法人实体 | 高信任度生产 |
| 4 | 扩展验证(EV) | 手动审查+法律 | 受监管行业 |
证据记录
每次工具调用(允许或拒绝)都会产生一个证据记录:
from capiscio_mcp import guard, GuardError
@guard(min_trust_level=2)
async def sensitive_operation(data: dict) -> dict:
pass
try:
result = await sensitive_operation(data={"key": "value"})
except GuardError as e:
# Evidence logged even on denial
print(f"Denied: {e.reason}")
print(f"Evidence ID: {e.evidence_id}") # For audit trail证据包括:
- 工具名称和参数哈希(不是原始参数——PII安全)
- 呼叫者身份(代理DID、徽章JTI、身份验证级别)
- 决定和理由
- 时间戳和唯一证据ID
配置参考
GuardConfig
from capiscio_mcp import GuardConfig
config = GuardConfig(
min_trust_level=2, # Minimum trust level (0-4)
accept_level_zero=False, # Accept self-signed badges?
trusted_issuers=[ # List of trusted issuer DIDs
"did:web:registry.capisc.io",
],
allowed_tools=[ # Glob patterns for allowed tools
"read_*",
"list_*",
],
require_badge=True, # Deny anonymous/API key access
policy_version="v1.0", # Policy version for tracking
)验证配置。
from capiscio_mcp import VerifyConfig
config = VerifyConfig(
trusted_issuers=[...], # Trusted issuer DIDs
min_trust_level=2, # Minimum required level
accept_level_zero=False, # Accept self-signed servers?
offline_mode=False, # Skip revocation checks?
skip_origin_binding=False, # Skip host/path binding?
)环境变量
| 变量 | 描述 | 默认值 |
|---|---|---|
CAPISCIO_SERVER_ID | 服务器UUID(用于 MCPServerIdentity) | — |
CAPISCIO_API_KEY | 注册表API项(用于 MCPServerIdentity) | — |
CAPISCIO_SERVER_URL | 注册表服务器URL | https://registry.capisc.io |
CAPISCIO_SERVER_DOMAIN | 徽章发放域 | (来源于服务器URL) |
CAPISCIO_SERVER_PRIVATE_KEY_PEM | PEM编码的Ed25519私钥(临时envs) | -- |
CAPISCIO_CORE_ADDR | 外部核心地址 | (嵌入式模式) |
CAPISCIO_SERVER_ORIGIN | 防护服务器来源 | (自动检测) |
CAPISCIO_LOG_LEVEL | 记录冗长 | info |
API 参考
防护装置(RFC-006)
guard(config=None, min_trust_level=None, tool_name=None)--异步装饰器guard_sync(...)--同步装饰器evaluate_tool_access(tool_name, params, credential, config)-低级APIcompute_params_hash(params)--确定性参数散列GuardConfig--配置数据类GuardResult--评估结果数据类GuardError--拒绝访问的异常
服务器(RFC-007)
verify_server(server_did, server_badge, transport_origin, endpoint_path, config)--异步验证verify_server_sync(...)--同步验证verify_server_strict(...)--在任何验证失败时引发ServerVerifyErrorparse_http_headers(headers)--从HTTP标头中提取身份信息parse_jsonrpc_meta(meta)--从MCP \_数据中提取标识VerifyConfig--配置数据类VerifyResult--验证结果数据类ServerVerifyError--验证失败的例外情况
注册(服务器标识)
generate_server_keypair(key_id, output_dir)--生成Ed25519密钥对generate_server_keypair_sync(...)--同步版本register_server_identity(server_id, api_key, did, public_key, ca_url)--在注册表中注册DIDregister_server_identity_sync(...)--同步版本setup_server_identity(server_id, api_key, ca_url, output_dir, key_id)--组合设置setup_server_identity_sync(...)--同步版本RegistrationError--注册失败的例外情况KeyGenerationError--密钥生成失败的例外情况
类型
Decision--允许/拒绝AuthLevel-ANONYMOUS/API_KEY/徽章DenyReason--列举拒绝理由TrustLevel--根据RFC-002,信任级别为0-4ServerState--已验证_原则/声明_原则/未验证_原产地ServerErrorCode--验证错误代码枚举
MCP SDK集成(可选)
需要 pip install capiscio-mcp[mcp]:
CapiscioMCPServer.connect()--一行代码:从env加载标识并创建服务器CapiscioMCPServer(name, did, badge, ...)--具有信任强制的FastMCP包装器CapiscioMCPServer.tool(min_trust_level=...)--防护工具装饰器CapiscioMCPServer.run(transport="stdio")--运行服务器CapiscioMCPClient(command, args, ...)--stdio传输客户端\*CapiscioMCPClient.call_tool(name, args)--调用服务器上的工具CapiscioMCPClient.list_tools()--列出可用工具
\*注意:中的服务器身份验证 CapiscioMCPClient 需要MCP SDK支持 _meta 初始化响应中的传递。这还不可用,所以 min_trust_level 和 fail_on_unverified 当前未强制执行参数。通过以下方式执行服务器端信任 @server.tool(min_trust_level=...) 完全工作。
文档
发展
# Clone repository
git clone https://github.com/capiscio/capiscio-mcp-python.git
cd capiscio-mcp-python
# Install development dependencies
pip install -e ".[dev]"
# Run tests
pytest -v
# Run tests with coverage
pytest --cov=capiscio_mcp --cov-report=html
# Type checking
mypy capiscio_mcp
# Linting
ruff check capiscio_mcp相关套餐
| 软件包 | 功能 | 安装 |
|---|---|---|
| 特工警卫 | A2A代理的运行时信任验证 | pip install capiscio-sdk |
| CapiscIO命令行界面 | CI/CD管道的代理验证 | pip install capiscio |
许可证
Apache许可证2.0
贡献
看 贡献.md 作为指导方针。
