capiscio/a2a-demos
GitHub: capiscio/a2a-demos
一套展示AI代理信任徽章、身份验证和工具级授权的演示项目，用于保障A2A和MCP协议通信安全。
Stars: 2 | Forks: 0
# CapiscIO 演示
## 演示概览
| 演示 | 展示内容 | 时长 | 快速开始 |
|------|---------------|------|-------------|
| **[演示一：从零到强制执行](#demo-one--zero-to-enforcement)** | `@guard` 装饰器、信任等级、徽章访问控制 | 5 分钟 | `cd demo-one && ./setup.sh` |
| **[演示二：策略即代码](#demo-two--policy-as-code)** | 运行时策略变更改变强制执行效果 — 无需代码部署 | 10 分钟 | `cd demo-two && ./setup.sh` |
| **[MCP Guard 演示](#mcp-guard-demo)** | 服务器身份、每工具信任、客户端验证 | 5 分钟 | `cd mcp-demo && docker compose up` |
| **[Agent Guard 演示](#agent-guard-demos)** | 3 个框架代理带 DID、徽章、实时事件 | 15 分钟 | `./scripts/setup.sh` |
**初次接触 CapiscIO？** 从演示一开始 — 只需 5 分钟即可了解核心概念。
## 前置条件
- Python 3.11+
- 免费 CapiscIO 账户 — 在 [app.capisc.io](https://app.capisc.io) 注册
- 来自仪表板 → 设置 → API 密钥的 API 密钥
- 在仪表板中注册的 MCP 服务器（用于 demo-one 和 demo-two）
## 演示一 — 从零到强制执行
**"5 分钟从零到受信任强制的 MCP 工具。"**
一个具有三个不同信任等级的 MCP 工具的 MCP 服务器。受信任的代理（带有徽章）可以调用受限工具；不受信任的代理（无徽章）将被拒绝。
### 你将看到的内容
| 场景 | 代理 | 工具 | 信任等级 | 结果 |
|----------|-------|------|-------------|--------|
| 1 | 受信任（DV 徽章） | `get_price` | 0（开放） | 允许 |
| 2 | 受信任（DV 徽章） | `place_order` | 2（DV+） | 允许 |
| 3 | 不受信任（无徽章） | `get_price` | 0（开放） | 允许 |
| 4 | 不受信任（无徽章） | `place_order` | 2（DV+） | **拒绝** |
### 设置
```
cd demo-one
./setup.sh # Creates venv, installs deps, downloads binary
cp .env.example .env # Fill in your API key + server ID
```
### 运行
```
source .venv/bin/activate
python run_demo.py
```
### 关键代码
**服务器** — 每个工具一个装饰器：
```
@server.tool(min_trust_level=0)
async def get_price(sku: str) -> str: ...
@server.tool(min_trust_level=2)
async def place_order(sku: str, quantity: int) -> str: ...
@server.tool(min_trust_level=4)
async def cancel_all_orders() -> str: ...
```
**代理** — 一行代码连接：
```
identity = CapiscIO.connect(api_key=..., auto_badge=True)
```
### 文件
```
demo-one/
├── server/main.py # MCP server with 3 guarded tools
├── agents/
│ ├── trusted_agent.py # Badged agent (auto_badge=True)
│ └── untrusted_agent.py # No-badge agent (auto_badge=False)
├── run_demo.py # Orchestrator: 4 scenarios
├── setup.sh # Environment setup + binary download
├── .env.example # Credential template
└── requirements.txt
```
## 演示二 — 策略即代码
**"相同代码，三种不同强制执行结果 — 由策略改变，而非部署。"**
展示组织级策略如何在运行时改变信任强制执行。演示者在各阶段之间在仪表板中切换策略；相同的代理和服务器会产生不同的允许/拒绝结果。
### 三个阶段
**阶段 1 — 基线**（信任等级按代码设置）
| 代理 | get_price | place_order |
|-------|-----------|-------------|
| 受信任（DV） | 允许 | 允许 |
| 不受信任 | 允许 | **拒绝** |
**阶段 2 — 锁定**（全局最小值提升至 EV）
| 代理 | get_price | place_order |
|-------|-----------|-------------|
| 受信任（DV） | **拒绝** | **拒绝** |
| 不受信任 | **拒绝** | **拒绝** |
**阶段 3 — 选择性**（get_price 覆盖为需要 DV）
| 代理 | get_price | place_order |
|-------|-----------|-------------|
| 受信任（DV） | 允许 | 允许 |
| 不受信任 | **拒绝** | **拒绝** |
### 设置
```
cd demo-two
./setup.sh
cp .env.example .env # Fill in API key, server ID, org ID, admin JWT
```
创建三个策略提案：
```
source .venv/bin/activate
python scripts/setup_policies.py
```
### 运行
```
python run_demo.py
```
脚本在各阶段之间暂停，以便你可以在仪表板中切换策略。
### 策略文件
```
# policies/lockdown.yaml — 紧急响应
version: "1"
min_trust_level: "EV"
```
```
# policies/selective.yaml — 按工具覆盖
version: "1"
mcp_tools:
- tool: "get_price"
min_trust_level: "DV"
```
### 文件
```
demo-two/
├── policies/
│ ├── baseline.yaml # Default enforcement
│ ├── lockdown.yaml # Global min = EV (deny all)
│ └── selective.yaml # get_price overridden to DV
├── scripts/
│ └── setup_policies.py # Creates policy proposals via admin JWT
├── server/main.py # Same MCP server as demo-one
├── agents/ # Same agents as demo-one
├── run_demo.py # Interactive 3-phase orchestrator
├── setup.sh
├── .env.example
└── requirements.txt
```
## MCP Guard 演示
**"MCP 服务器的 Let's Encrypt"** — 自动加密身份、信任徽章和每工具访问控制。
| 功能 | 描述 |
|---------|-------------|
| `MCPServerIdentity.connect()` | 一行代码：生成密钥、注册 DID、获取徽章 |
| 服务器身份在 `_meta` 中 | 每个 `initialize` 响应都携带服务器的 DID + 徽章 |
| 每工具信任等级 | `@server.tool(min_trust_level=N)` — 例如 `list_files`=0, `read_file`=2, `write_file`=3 |
| 客户端验证 | 客户端在调用工具之前验证服务器 DID + 徽章 |
| 自动续期 | `ServerBadgeKeeper` 在徽章过期前续期 |
### 快速开始
```
cd mcp-demo
cp .env.example .env # Set CAPISCIO_SERVER_ID + CAPISCIO_API_KEY
docker compose up --build # Starts registry, MCP server, and client
```
**→ 完整设置、架构和预期输出：[`mcp-demo/README.md`](mcp-demo/README.md)**
## Agent Guard 演示
运行 **3 个 AI 代理**，分别使用不同框架构建，均由 CapiscIO 信任徽章保护：
- **LangChain** — 带工具调用的研究代理（端口 8001）
- **CrewAI** — 用于创意任务的多代理团队（端口 8002）
- **LangGraph** — 带复杂工作流程的有状态代理（端口 8003）
所有代理都使用 `CapiscIO.connect()` 获取加密身份（DID）、注册到注册表，并参与受信任的代理间通信。通过 [CapiscIO 仪表板](https://app.capisc.io) 实时查看它们的**事件日志**。
### 快速开始
### 前置条件
- Python 3.11+（3.14+ 可用但会显示弃用警告）
- OpenAI API 密钥（或兼容的 LLM）
- 免费 CapiscIO 账户 — 在 [app.capisc.io](https://app.capisc.io) 注册
### 1. 设置代理环境
```
cd a2a-demos
./scripts/setup.sh # Creates per-agent .venvs, installs deps + shared module
cp .env.example .env
```
### 2. 配置环境
使用你的凭据编辑 `.env`：
```
OPENAI_API_KEY=sk-your-openai-key
OPENAI_MODEL=gpt-4o-mini
CAPISCIO_SERVER_URL=https://registry.capisc.io
CAPISCIO_API_KEY=sk_live_your_api_key_here
SECURITY_MODE=ca
```
从 [app.capisc.io](/.capiscio/keys/`
2. **派生 `did:key` URI** — 从公钥（RFC-002 §6.1）
3. **注册到注册表** — 通过 `/v1/sdk/agents` 创建代理记录
4. **修补 DID + 公钥** — 将加密身份链接到代理
5. **激活代理** — 将状态设置为"active"
6. **启动 BadgeKeeper** — 后台线程自动续信任徽章
7. **提供 A2A 端点** — 代理卡在 `/.well-known/agent.json`，任务在 `/tasks/send`
## 项目结构
```
a2a-demos/
├── demo-one/ # Zero to Enforcement (5 min)
│ ├── server/main.py # MCP server with 3 guarded tools
│ ├── agents/ # Trusted + untrusted agents
│ ├── run_demo.py # 4-scenario orchestrator
│ └── setup.sh # One-command setup
├── demo-two/ # Policy as Code (10 min)
│ ├── policies/ # 3 YAML policy files
│ ├── scripts/setup_policies.py # Policy creation via admin JWT
│ ├── run_demo.py # Interactive 3-phase orchestrator
│ └── setup.sh
├── mcp-demo/ # MCP Guard demo (Docker)
│ ├── server/main.py # Guarded MCP filesystem server
│ ├── client/main.py # Client with server verification
│ ├── docker-compose.yml # Full stack orchestration
│ └── README.md # Detailed MCP demo docs
├── agents/
│ ├── langchain-agent/ # LangChain research agent (port 8001)
│ ├── crewai-agent/ # CrewAI multi-agent crew (port 8002)
│ └── langgraph-agent/ # LangGraph stateful agent (port 8003)
├── scripts/
│ ├── setup.sh # Create venvs, install deps
│ ├── run-agents.sh # Launch all 3 agents (tmux or manual)
│ └── demo_driver.py # Send A2A tasks between agents
├── shared/
│ └── capiscio_events/ # Shared event emission module
├── .env.example # Environment template
└── README.md
```
## 架构
```
┌─────────────────────────────────────────────────────────┐
│ CapiscIO Registry (registry.capisc.io) │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │ Badge CA │ │ Events │ │ Agent Registry │ │
│ │ /v1/badge│ │ /v1/events│ │ /v1/sdk/agents │ │
│ └──────────┘ └──────────┘ └──────────────────┘ │
└───────────┬─────────────────────────┬───────────────────┘
│ │
┌──────┴──────┐ ┌──────────┴──────────┐
│Agent Guard │ │ MCP Guard │
├─────────────┤ ├─────────────────────┤
│ │ │ │
│ LangChain │ │ MCP Server │
│ :8001 │ │ MCPServerIdentity │
│ │ │ .connect() │
│ CrewAI │ │ + per-tool trust │
│ :8002 │ │ │ │
│ │ │ stdio│transport │
│ LangGraph │ │ ▼ │
│ :8003 │ │ MCP Client │
│ │ │ verifies server │
│ A2A Proto │ │ DID + badge │
└─────────────┘ └─────────────────────┘
```
每个代理获得自己的加密身份（DID）和密钥对。徽章由 CapiscIO 注册表 CA 签名。
## 安全配置
通过环境变量控制徽章强制执行：
| 变量 | 默认值 | 描述 |
|----------|---------|-------------|
| `SECURITY_MODE` | `ca` | `ca` 为 CA 签名徽章，`dev` 为自签名 |
| `CAPISCIO_REQUIRE_SIGNATURES` | `false` | 传入 A2A 请求需要有效徽章 |
| `CAPISCIO_FAIL_MODE` | `block` | 安全失败时的操作：`block`、`monitor` 或 `log` |
| `CAPISCIO_MIN_TRUST_LEVEL` | `0` | 所需最小信任等级（0-3） |
| `CAPISCIO_RATE_LIMITING` | `true` | 启用速率限制 |
| `CAPISCIO_RATE_LIMIT_RPM` | `60` | 每分钟请求限制 |
严格配置示例：
```
CAPISCIO_REQUIRE_SIGNATURES=true
CAPISCIO_FAIL_MODE=block
CAPISCIO_MIN_TRUST_LEVEL=1
```
## 事件类型
在仪表板中可见的事件：
| 事件类型 | 描述 |
|-------------|-------------|
| `agent.started` | 代理已初始化并就绪 |
| `badge.requested` | 已向 CA 请求徽章 |
| `badge.renewed` | 徽章由 keeper 自动续期 |
| `task.started` | A2A 任务执行开始 |
| `task.completed` | 任务成功完成 |
| `a2a.request` | 发出出站 A2A 调用 |
| `a2a.response` | 收到 A2A 响应 |
| `error` | 出错 |
## 快速测试
启动代理后，验证一切正常：
```
# 1. 检查所有代理是否健康
curl -s http://localhost:8001/health # LangChain
curl -s http://localhost:8002/health # CrewAI
curl -s http://localhost:8003/health # LangGraph
# 2. 发现代理能力
python scripts/demo_driver.py --discover
# 3. 测试单个代理（快速 - 无需 LLM 调用）
python scripts/demo_driver.py --agent langgraph --task "My login is broken"
# 预期输出：
# ✅ 任务在 0.0 秒内完成
# 📋 响应：
# 我理解您遇到了技术问题...
```
## 🔧 故障排除
| 问题 | 原因 | 修复 |
|---------|-------|-----|
| `RuntimeError: capiscio binary not found` | 核心二进制文件不可用 | SDK 首次运行时自动下载；检查网络连接 |
| 代理启动时 `ConnectionRefusedError` | 注册表不可达 | 检查 `.env` 中的 `CAPISCIO_SERVER_URL` 和网络连接 |
| `Port 8001 already in use` | 之前的代理仍在运行 | `lsof -ti:8001 \| xargs kill` |
| 代理启动但仪表板中无事件 | API 密钥或服务器 URL 错误 | 验证 `.env` 中的 `CAPISCIO_API_KEY` 和 `CAPISCIO_SERVER_URL` |
| `OPENAI_API_KEY not set` | 缺少 `.env` | `cp .env.example .env` 并填写密钥 |
| `ModuleNotFoundError: capiscio_sdk` | SDK 未在 venv 中安装 | `source .venv/bin/activate && pip install capiscio-sdk` |
| Pydantic V1 弃用警告 | 使用 Python 3.14+ | 可以忽略；功能仍然正常 |
## 📚 了解更多
- [CapiscIO 文档](https://docs.capisc.io)
- [A2A 协议规范](https://github.com/google/a2a)
- [RFC-002：信任徽章规范](https://github.com/capiscio/capiscio-rfcs)
- [Python SDK 参考](https://docs.capisc.io/sdk/python)
## 📄 许可证
MIT — 参见 [LICENSE](LICENSE)
标签：A2A协议, AI代理, API安全, Badge认证, DID, Docker, JSON输出, MCP, Python, Streamlit, 代理安全, 信任机制, 去中心化标识符, 安全防御评估, 安全防护, 工具授权, 微服务安全, 无后门, 策略即代码, 聊天机器人安全, 访问控制, 请求拦截, 运行时策略, 逆向工具, 零信任架构