Dynamic MCP Poc

MCP网关——经过身份验证的动态工具发现

概念证明演示了一个经过身份验证的MCP（模型上下文协议）网关，该网关具有Keycloak、RFC 8693令牌交换和动态会话工具发现功能。

这有什么作用

AI代理连接到单个MCP网关。网关允许代理根据需要发现和激活工具服务器。每一跳都经过身份验证——用户通过Keycloak登录，网关将其令牌交换为服务器特定的令牌，每个MCP服务器验证其自己的作用域令牌。

User (browser)
  │
  │  Keycloak OIDC login (PKCE)
  ▼
Web Frontend + ADK Agent (:8000)
  │
  │  Bearer token (aud: mcp-gateway)
  ▼
MCP Gateway (:8010)
  │
  │  Token exchange (RFC 8693)
  ▼
MCP Servers (:8011, :8012)
  Each validates its own scoped token (aud: mcp-weather, mcp-calculator)

主要特点：

• 动态工具发现 --客服电话 search_servers / enable_server 在运行时
• 每会话隔离 --在某个会话中激活的服务器在另一个会话中不可见
• 代币兑换 --网关通过Keycloak将用户令牌交换为服务器特定的令牌
• 按用户访问控制 --Keycloak角色决定每个用户可以访问哪些服务器
• 无需更改代码即可添加服务器 --只需添加一个YAML条目和一个Keycloak客户端

先决条件

• Python 3.13+
• Docker（用于Keycloak）
• Node.js 18+（用于BDD测试）
• LLM API密钥（Vertex AI、Google AI Studio或OpenAI）

快速开始

1.克隆并设置Python环境

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

2.配置环境

cp .env.example .env

编辑 .env --按键设置为 MODEL_NAME:

# Vertex AI (default, requires gcloud auth)
MODEL_NAME=vertex_ai/gemini-2.5-flash

# Or Google AI Studio (requires API key)
MODEL_NAME=gemini/gemini-2.0-flash
GOOGLE_API_KEY=your-key-here

3.开始一切

make up

这将按顺序启动所有服务：Keycloak、配置令牌交换权限、MCP服务器、网关和web UI。它等待每个组件准备就绪，然后再开始下一个组件。

4.打开web UI

首选 http://localhost:8000 然后单击 使用Keycloak登录.

测试用户：

登录后，您将被重定向到ADK聊天UI。让客服找到并使用可用的工具。

发出命令

单独运行服务

如果您希望在单独的端子中启动组件：

# Terminal 1: Keycloak
docker compose up -d
# Wait for Keycloak to be ready (~30s), then configure permissions:
bash keycloak/setup-permissions.sh

# Terminal 2: Weather server
source .venv/bin/activate
python servers/weather_server.py

# Terminal 3: Calculator server
source .venv/bin/activate
python servers/calculator_server.py

# Terminal 4: Gateway
source .venv/bin/activate
python gateway/server.py

# Terminal 5: Web UI
source .venv/bin/activate
python agent/web.py

端口映射

运作原理

1. 用户通过Keycloak登录（授权码+PKCE）
2. 访问令牌（受众： mcp-gateway)被注入到代理的MCP呼叫中
3. 客服电话 search_servers() 发现可用服务器
4. 客服电话 enable_server("weather") --网关：

- 检查用户的角色(access:weather) - 将令牌交换为一个范围为 mcp-weather - 连接到天气服务器，发现其工具 - 在网关上注册代理工具

1. 客服电话 get_weather("Warsaw") --网关：

- 验证此会话中是否启用了服务器 - 再次将令牌交换为新的服务器范围令牌 - 将呼叫转发到天气服务器 - 返回结果

每个MCP会话都有自己的一组启用的服务器。在一个浏览器选项卡中激活的工具不会出现在另一个选项卡中。

运行BDD测试

该项目使用Cucumber.js和TypeScript对正在运行的服务进行黑盒BDD测试。

# Install test dependencies (once)
make test-install

# Start all services
make up

# Run tests
make test

项目结构

MCPTest/
├── gateway/
│   ├── server.py              # MCP gateway with auth + token exchange + per-session state
│   └── servers.yaml           # Server registry (name, URL, audience, role)
├── servers/
│   ├── weather_server.py      # Weather MCP server (JWT auth)
│   └── calculator_server.py   # Calculator MCP server (JWT auth)
├── agent/
│   ├── __init__.py            # Exports root_agent
│   ├── main.py                # ADK agent with MCP toolset + header_provider
│   └── web.py                 # FastAPI: Keycloak login + ADK Web UI
├── keycloak/
│   ├── mcp-poc-realm.json     # Realm config (clients, scopes, mappers, users)
│   └── setup-permissions.sh   # Configures V1 fine-grained token exchange permissions
├── tests/bdd/                 # Cucumber.js BDD test suite
│   ├── features/              # Gherkin scenarios
│   ├── steps/                 # TypeScript step definitions
│   └── support/               # World class
├── doc/
│   ├── ARCHITECTURE.md        # Architecture overview
│   ├── IMPLEMENTATION.md      # Implementation details
│   └── IMPLEMENTATION-BDD.md  # BDD test approach
├── docker-compose.yml         # Keycloak container
├── Makefile                   # Service orchestration
├── requirements.txt           # Python dependencies
└── .env.example               # Environment template

添加新的MCP服务器

1. 钥匙斗篷 --添加仅承载客户端(mcp-)，一个受众范围，并将其分配给 mcp-gateway
2. 服务器代码 --创建 servers/_server.py 随着 JWTVerifier(audience="mcp-")
3. 网关配置 --向添加条目 gateway/servers.yaml
4. 重启 网关——网关或代理中不需要更改代码

看 doc/IMPLEMENTATION.md 第10节了解全部细节。

文档

• 建筑 --组件设计、信任边界、令牌流
• 实施 --代码结构、Keycloak设置、关键API
• BDD测试 --测试方法、场景目录

限制（PoC）

• 仅内存状态--重新启动时丢失
• 无令牌刷新——令牌过期后（1小时）用户必须重新登录
• 通过全局变量进行单用户web会话
• 无TLS——所有流量都是明文HTTP