AIP电子邮件服务——示例应用
一个最小的Python演示,显示 代理身份协议(AIP) 在行动中——两者都有 CLI演示 和一个 web用户界面.
它展示了什么
| 概念 | 地点 |
|---|---|
| 第1层——身份 --发出已签名的代理身份验证令牌(AAT) | auth.py |
| 第2层——执行 --在每个请求上验证AAT,执行策略 | mcp_server.py, webapp.py |
| 基于角色的数据隔离 --用户只能看到自己的电子邮件;管理员看到全部 | mcp_server.py, webapp.py |
| 能力检查 --令牌带有明确的能力声明 | auth.py |
不可变审计日志 --每个允许/拒绝都附加到 audit.jsonl | mcp_server.py, webapp.py |
| 运输不可知论 --AAT通过stdio、HTTP承载和HTTP上的MCP工作 | auth.py |
| MCP标准传输 --代理接口支持JSON-RPC 2.0 | mcp_server.py |
| 基于HTTP/SSE的MCP --Claude桌面/光标通过URL连接,无子进程 | webapp.py |
| 单独的身份验证路径 --人类浏览器会话与代理AAT是不同的 | webapp.py |
建筑
auth.py + data.py
(shared, unchanged)
┌──────┴──────┐
│ │
mcp_server.py webapp.py
(AIP Layer 2 for ┌── Browser routes ──────────────────────┐
MCP stdio agents) │ plain session {user_id, role} │
│ │ no AAT — humans are not agents │
main.py ├── JSON API routes (/api/*) ─────────────┤
(CLI demo) │ Authorization: Bearer │
│ AIP Layer 2 enforcement │
├── MCP over HTTP/SSE (/aip-playground-mcp)┤
│ Claude Desktop / Cursor connect here │
│ authenticate tool → issues AAT (L1) │
│ list_* tools → enforce AAT (L2) │
└─────────────────────────────────────────┘关键见解: auth.py 与交通无关。相同 validate_aat() / check_capability() / check_role() 功能在所有三个方面执行策略 代理路径:stdio MCP、HTTP承载API和MCP-over-HTTP/SSE。
文件
.
├── auth.py AIP Layer 1: AAT issuance & Layer 2: validation helpers
├── data.py In-memory email store and user registry
├── mcp_server.py AIP-aware MCP server (stdio JSON-RPC 2.0 transport)
├── mcp_server_plain.py Plain MCP server — no enforcement (baseline / aip-go target)
├── main.py CLI demo — runs all four scenarios
├── webapp.py FastAPI web UI + JSON API + MCP over HTTP/SSE
├── requirements.txt Web UI dependencies
├── requirements-dev.txt Test dependencies (pytest, httpx)
├── tutorials.md Step-by-step walkthroughs
├── tests/ Test suite (94 tests)
│ ├── test_auth.py AIP Layer 1 + Layer 2 unit tests
│ ├── test_data.py Data store unit tests
│ ├── test_mcp.py stdio MCP server + plain server integration tests
│ └── test_webapp.py Browser routes, JSON API, MCP dispatch tests
└── audit.jsonl Created at runtime; one JSON line per event快速开始
Web用户界面
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
uvicorn webapp:app --reload --port 8000打开 http://localhost:8000 并使用任何模拟账户登录:
| 用户名 | 密码 | 角色 |
|---|---|---|
alice | alice123 | 用户--只看到自己的收件箱 |
bob | bob123 | 用户--只看到自己的收件箱 |
admin | admin123 | 管理员--查看所有收件箱+审核日志 |
CLI演示
python3 main.py # run all four scenarios
python3 main.py --demo user # user agent: own inbox only
python3 main.py --demo admin # admin agent: all inboxes
python3 main.py --demo forbidden # user tries admin tool → denied
python3 main.py --demo invalid # forged token → deniedCLI不需要依赖项——简单的Python 3.9+。
Web UI页面和路由
浏览器路由
| 路线 | 访问 | 描述 |
|---|---|---|
/login | 公共 | 登录表单 |
/inbox | 经过身份验证的用户 | 仅限自己的电子邮件(AIP数据隔离) |
/admin | 管理员角色 | 所有用户的所有电子邮件 |
/audit | 管理员角色 | 实时审核日志--所有传输 |
代理路线
| 路线 | 授权 | 描述 |
|---|---|---|
GET /api/emails/mine | Bearer — read:own_emails | JSON API-呼叫者自己的电子邮件 |
GET /api/emails/all | Bearer — read:all_emails +admin | JSON API-所有电子邮件 |
GET /aip-playground-mcp | 无(SSE握手) | HTTP/SSE上的MCP——克劳德桌面/光标入口点 |
POST /aip-playground-mcp/messages?sessionId= | 每工具AAT | MCP JSON-RPC消息 |
身份验证路径
webapp.py 有三个完全独立的身份验证路径:
1.浏览器(人)-会话cookie
POST /login→authenticate_user()验证凭据→ 商店{user_id, role}在签名的会话cookie中。未签发AAT。- 每个浏览器路由读取
session["user_id"],检查user["role"]直接。 - 审核记录:
actor = user_id,transport = "http".
2.JSON API(代理)-承载令牌
- 每
/api/*路线读取Authorization: Bearer→validate_aat()→check_capability()/check_role(). - 审核记录:
actor = agent_id,transport = "http".
3.基于HTTP/SSE的MCP(代理)——每个工具AAT
- 客户端打开
GET /aip-playground-mcp→ 接收带有消息端点URL的SSE流。 - 客户端POST JSON-RPC到
/aip-playground-mcp/messages?sessionId=. authenticate工具:验证凭据、调用issue_aat()(第1层)返回签名的AAT。list_my_emails/list_all_emails:验证AAT(第2层),执行能力/角色。- 审核记录:
actor = agent_id,transport = "mcp-http".
AAT索赔
AAT由 代理 或注册表来验证代理(不是人类)的身份——例如,通过 main.py 或客户代理来电 issue_aat()。人类浏览器登录可以 不 生成AAT。
{
"iss": "aip-sample-issuer",
"sub": "alice",
"iat": 1736500000,
"exp": 1736503600,
"jti": "a3f8d1c2",
"agent_id": "email-assistant-v1",
"role": "user",
"capabilities": ["read:own_emails"]
}| 索赔 | 含义 |
|---|---|
iss | 代币发行人(AIP注册表) |
sub | 代理所代表的人类用户 |
agent_id | 的唯一标识符 代理 (不是人类) |
role | 权限级别(user / admin) |
capabilities | 允许操作的明确列表 |
jti | 令牌ID——用于吊销检查 |
审核日志示例
浏览器和API事件共享相同 audit.jsonl 文件,以 transport 和 actor:
{"ts":"...","event":"login", "actor":"alice", "action":"login", "outcome":"allow","transport":"http"}
{"ts":"...","event":"page_view", "actor":"alice", "action":"inbox", "outcome":"allow","transport":"http"}
{"ts":"...","event":"page_view", "actor":"alice", "action":"admin", "outcome":"deny", "transport":"http"}
{"ts":"...","event":"tool_call", "actor":"claude-agent", "action":"authenticate", "outcome":"allow","transport":"mcp-http"}
{"ts":"...","event":"tool_call", "actor":"claude-agent", "action":"list_my_emails", "outcome":"allow","transport":"mcp-http"}
{"ts":"...","event":"tool_call", "actor":"email-assistant", "action":"list_my_emails", "outcome":"allow","transport":"mcp"}
{"ts":"...","event":"api_call", "actor":"email-assistant", "action":"api:list_my_emails","outcome":"allow","transport":"http"}transport | 来源 |
|---|---|
"http" | 浏览器页面浏览量和直接 /api/* 电话 |
"mcp" | stdio MCP服务器(mcp_server.py) |
"mcp-http" | 基于HTTP/SSE的MCP(/aip-playground-mcp) |
TODO:指出代理正在执行基于浏览器的操作的时间点。
MCP服务器→ web服务器:AAT与会话
这是一个关键的区别——浏览器会话和代理AAT是 完全分开.
人类没有AATs。代理没有会话。
| 浏览器(人工) | JSON API代理 | MCP-over-HTTP代理 | |
|---|---|---|---|
| 入口点 | /login 形式 | /api/* | GET /aip-playground-mcp |
| 身份载体 | 会话cookie {user_id, role} | Authorization: Bearer | AAT 通过 authenticate 工具 |
issue_aat() 打电话? | 从不 | 由外部代理 | 由 authenticate 工具(第1层) |
| 验证人: | _session_user() | _bearer_claims() + validate_aat() | validate_aat() 每次工具调用 |
| 审计运输 | "http" | "http" | "mcp-http" |
Browser → session cookie → _session_user() → user dict
stdio agent → aat= tool argument → mcp_server.py → validate_aat()
API client → Authorization: Bearer → _bearer_claims() → validate_aat()
MCP-HTTP agent → authenticate tool → issue_aat() [Layer 1]
→ list_* tools + aat → validate_aat() [Layer 2]MCP服务器调用web服务器
mcp_server.py 具有两种操作模式。在 微服务模式 (WEBAPP_BASE_URL set),它将代理的AAT作为Bearer令牌转发到web服务器的JSON API,并且不读取 data.py 直接。在 单机模式 (没有 WEBAPP_BASE_URL),它进口 data.py 直接——这就是 main.py 在没有运行web服务器的情况下使用。
AI Agent
│ tools/call + aat
▼
mcp_server.py
├── validate_aat() local fast-fail before any HTTP
├── check_capability()
└── GET /api/emails/mine HTTP, Authorization: Bearer
│
▼
webapp.py
├── validate_aat() re-enforces independently
├── check_capability()
└── data.py single authoritative data source执法运行 两次 --本地在MCP服务器中(快速故障),再次在web服务器中(权威)。web服务器不信任任何调用者。这是微服务后端的正确模式:一个数据源,多个前端,每个前端独立执行相同的策略。
JSON API路由
| 路由 | 标题 | 所需功能 | 角色 | 返回 |
|---|---|---|---|---|
GET /api/emails/mine | Authorization: Bearer | read:own_emails | 任何 | 来电者自己的电子邮件 |
GET /api/emails/all | Authorization: Bearer | read:all_emails | admin | 所有电子邮件 |
响应: 401 (丢失/无效令牌), 403 (能力或角色不足), 200 JSON成功。
一起跑步
# Terminal 1 — web server
source .venv/bin/activate
uvicorn webapp:app --port 8000
# Terminal 2 — MCP server, pointed at the web server
export WEBAPP_BASE_URL=http://localhost:8000
python3 mcp_server.py集 WEBAPP_BASE_URL 将MCP服务器指向任何已部署的实例 webapp.py.
______________________________________________________________________
与AIP规范的关系
此示例是Python中概念的说明 aip-v1alpha1.md.
Go引用代理 好了 包裹 *任何* MCP服务器,并从外部处理第2层执行。 这个演示将这两个层捆绑到一个Python进程中以保持 该示例自包含且易于理解。
看 实施.md 获取连接的分步指南 我去这个操场 mcp_server_plain.py 来自Cursor、Claude Desktop或 命令行。
