内容
# agentgate
**原生的智能代理网页协议。给AI代理一个结构化的API，而不是网站。**
---
当一个AI代理今天访问`foodpanda.com`时，它会下载HTML，这些HTML是为人类眼睛设计的——导航菜单、广告横幅、模态窗口——并烧毁上下文窗口和计算资源，试图将其解析为操作。`agentgate`从源头上解决了这个问题。
网站运营者只需一个下午就能将`agentgate`添加到他们的FastAPI后端。从那时起，任何发送`X-Agent-Request: true`的AI代理都会返回一个机器可读的清单，描述站点可以做什么以及如何做——没有HTML，没有爬虫，没有围绕DOM结构的提示工程。
```
代理 ──→ GET /.well-known/agent-manifest.json ──→ {意图, 流, 模式}
代理 ──→ POST /agent/flows/order_food/start ──→ {session_id, entry_step}
代理 ──→ POST /agent/flows/order_food/step/... ──→ {result, next_steps}
```
---
## 为什么不是MCP？
[模型上下文协议](https://modelcontextprotocol.io)（MCP）在其领域内表现出色：连接大型语言模型（LLM）到**本地或运营商控制环境中的工具和资源**——文件系统、数据库、内部API、开发工具。它是一个客户端协议，旨在扩展模型在一个会话内可以做的事情。
`agentgate`解决了一个不同的问题：**任何公共互联网上的网站都可以被代理原生调用，无需事先集成工作**。
| | MCP | agentgate |
|---|---|---|
| **存在位置** | 边车进程、标准输入/输出或SSE | 嵌入到现有的Web服务器中 |
| **谁配置它** | LLM客户端/操作员 | 网站操作员 |
| **发现** | 预配置 - 代理必须知道服务器存在 | 通用：`GET /.well-known/agent-manifest.json`在任何域名上 |
| **主要目的** | 扩展LLM工具访问（文件、数据库、API） | 使任何网站都可以被代理原生调用 |
| **多步骤工作流** | 手动 - 代理使用原始工具调用编排 | 内置：`server`和`client`编排模式 |
| **会话/状态** | 不是协议的一部分 | 一流：服务器在步骤之间持有状态 |
| **传输** | 标准输入/输出、SSE、HTTP（自定义） | 标准HTTP - 与任何`httpx`/`requests`客户端兼容 |
| **现有网站** | 需要单独的MCP服务器进程 | `app.mount(fastapi_app)` - 一行代码 |
| **代理兼容性** | 需要MCP感知客户端 | 任何HTTP客户端；不需要框架 |
| **开放Web规模** | 设计用于封闭、操作员控制的工具 | 设计用于开放Web - 在运行时发现任何域名 |
MCP和`agentgate`是**互补的，而不是竞争的**。代理可以使用MCP访问其自己的工具（内存、代码执行、日历），同时使用`agentgate`与外部网站（电子商务、预订、搜索）交互。
---
## 它是如何工作的
```
┌─────────────────────────────────────────────────────────────┐
│ 网站（FastAPI） │
│ │
│ agentgate.mount() ──添加──▶ /.well-known/agent-manifest │
│ /agent/intents/{name} │
│ /agent/flows/{name}/start │
│ /agent/flows/{name}/step/... │
│ │
│ 普通路由不受影响 - 浏览器看到无变化 │
└─────────────────────────────────────────────────────────────┘
▲
│ X-Agent-Request: true
│
┌─────────────────────┐
│ AI 代理 │
│ │
│ client.discover() │ ──▶ 获取清单，了解功能
│ site.intents.*() │ ──▶ 一次性结构化调用
│ site.flows.*.start()──▶ 启动有状态的多步骤工作流
│ session.step(...) │ ──▶ 继续工作流，服务器验证转换
└─────────────────────┘
```
1. **发现**：代理获取`/.well-known/agent-manifest.json`。清单列出了站点公开的每个意图（单步调用）和流程（多步骤工作流），以及输入和输出的完整JSON模式。不需要先前的知识。
2. **调用意图**：单目的操作（搜索、查找、翻译）- 一个请求，一个响应。
3. **执行流程**：多步骤过程（订购食品、预订航班、注册帐户）。两种模式：
- **服务器编排**：服务器持有会话状态并强制有效的步骤转换。代理不能跳过步骤或提前跳转 - 服务器验证所有操作。
- **客户端编排**：每个步骤都是独立的、无状态的端点。代理管理数据链接。适用于幂等或读取繁重的工作流。
---
## 安装
```bash
pip install agentgate
```
运行开发服务器：
```bash
pip install "agentgate[server]" # 包括uvicorn
```
**要求**：Python 3.10+，FastAPI，Pydantic v2，httpx
---
## 快速入门 - 服务器端
将`agentgate`添加到现有的FastAPI应用中，分为四个步骤。
### 1. 创建一个`AgentApp`
```python
from agentgate import AgentApp, AuthDef, step
from fastapi import FastAPI
agent_app = AgentApp(
title="MyShop",
description="E-commerce API for AI agents",
version="1.0",
auth=AuthDef(type="api_key", header="X-API-Key"),
)
```
### 2. 注册意图（单步功能）
装饰任何异步函数。输入和输出模式根据类型提示使用Pydantic自动生成。
```python
from pydantic import BaseModel
class Product(BaseModel):
id: str
name: str
price: float
@agent_app.intent("search_products", description="Search products by keyword")
async def search_products(query: str, max_results: int = 10) -> list[Product]:
# 现有业务逻辑
return db.search(query, limit=max_results)
```
### 3. 注册流程（多步骤工作流）
在类上使用`@agent_app.flow`，在方法上使用`@step`。
**服务器编排** - 服务器持有会话状态并强制步骤顺序：
```python
class CartItem(BaseModel):
product_id: str
quantity: int
class OrderConfirmation(BaseModel):
order_id: str
total: float
eta_minutes: int
@agent_app.flow("checkout", orchestration="server", description="Add to cart and place order")
class CheckoutFlow:
@step(entry=True, description="Add a product to the cart")
async def add_to_cart(self, product_id: str, quantity: int) -> dict:
return {"cart_id": cart.create(product_id, quantity)}
@step(after="add_to_cart", description="Confirm and place the order")
async def place_order(self, cart_id: str, address: str) -> OrderConfirmation:
order = orders.create(cart_id, address)
return OrderConfirmation(order_id=order.id, total=order.total, eta_minutes=30)
```
**客户端编排** - 每个步骤都是无状态和独立的：
```python
@agent_app.flow("track_order", orchestration="client", description="Track an existing order")
class TrackOrderFlow:
@step(entry=True, description="Initiate tracking for an order")
async def initiate(self, order_id: str) -> dict:
return {"tracking_id": tracking.start(order_id)}
@step(after="initiate", description="Poll for current status")
async def get_status(self, tracking_id: str) -> dict:
return {"status": tracking.get(tracking_id)}
```
### 4. 安装到FastAPI应用上
```python
fastapi_app = FastAPI(title="MyShop")
agent_app.mount(fastapi_app, base_url="https://myshop.com")
```
就是这样。您的现有路由保持不变。发送`X-Agent-Request: true`的代理获得代理API；普通浏览器看到的内容没有任何不同。
**启动服务器：**
```bash
uvicorn myapp:fastapi_app --reload
```
---
## 快速入门 - 代理/客户端
```python
import asyncio
from agentgate import AgentClient
async def main():
client = AgentClient(
agent_id="my-shopping-agent",
api_key="sk-...", # 根据清单认证类型自动注入
)
# 运行时发现功能 - 不需要事先了解API
site = await client.discover("https://myshop.com")
print(site)
# <DiscoveredSite 'MyShop' intents=['search_products'] flows=['checkout', 'track_order']>
# 调用单个意图
products = await site.intents.search_products(query="wireless headphones", max_results=5)
# 执行服务器编排的流程
session = await site.flows.checkout.start()
cart = await session.step("add_to_cart", product_id=products[0]["id"], quantity=1)
order = await session.step("place_order", cart_id=cart["cart_id"], address="Berlin, DE")
print(order) # {"order_id": "...", "total": 79.99, "eta_minutes": 30}
print(session.is_complete) # True
# 执行客户端编排的流程
tracking = await site.flows.track_order.steps.initiate(order_id=order["order_id"])
status = await site.flows.track_order.steps.get_status(tracking_id=tracking["tracking_id"])
print(status)
asyncio.run(main())
```
---
## 完整示例 - FoodPanda
`examples/`目录包含一个完整的服务器+客户端对：
```bash
# 终端1 - 启动服务器
uvicorn examples.foodpanda_server:fastapi_app --reload
# 终端2 - 运行代理
python examples/agent_client.py
```
[examples/foodpanda_server.py](examples/foodpanda_server.py) - 类似FoodPanda的后端，包含搜索意图、服务器编排的订购流程和客户端编排的跟踪流程。
[examples/agent_client.py](examples/agent_client.py) - 发现、调用意图并执行两种类型的流程的代理。
---
## 核心概念
### 清单
每个启用agentgate的站点都提供`GET /.well-known/agent-manifest.json`。这是站点与任何代理之间的契约。清单**根据类型注释自动生成** - 您不需要手动编写JSON模式。
```json
{
"agent_api_version": "1.0",
"name": "MyShop",
"description": "E-commerce API for AI agents",
"base_url": "https://myshop.com",
"auth": {
"type": "api_key",
"header": "X-API-Key"
},
"intents": [
{
"name": "search_products",
"description": "Search products by keyword",
"endpoint": "/agent/intents/search_products",
"method": "POST",
"input_schema": {
"type": "object",
"properties": {
"query": { "type": "string" },
"max_results": { "type": "integer" }
},
"required": ["query"]
},
"output_schema": { ... }
}
],
"flows": [
{
"name": "checkout",
"orchestration": "server",
"steps": [
{
"name": "add_to_cart",
"entry": true,
"next_steps": ["place_order"],
"input_schema": { ... },
"output_schema": { ... }
},
{
"name": "place_order",
"entry": false,
"next_steps": [],
"input_schema": { ... },
"output_schema": { ... }
}
]
}
]
}
```
### 意图
意图是一种单步、无状态的功能。意图用于查找、搜索、计算 - 任何接受输入并返回结果的操作。
```python
@agent_app.intent("convert_currency", description="Convert between currencies")
async def convert_currency(amount: float, from_currency: str, to_currency: str) -> dict:
rate = fx.get_rate(from_currency, to_currency)
return {"converted": amount * rate, "rate": rate}
```
代理调用它：
```
POST /agent/intents/convert_currency
{"amount": 100, "from_currency": "USD", "to_currency": "EUR"}
```
### 流程
流程是一种命名、有序的步骤序列。流程用于多步骤业务流程 - 预订、订购、注册、配置向导。
每个`@step`装饰器通过名称链接步骤：
- `entry=True` - 这是代理必须调用的第一个步骤
- `after="step_name"` - 此步骤在`step_name`完成后变为可用
**选择服务器和客户端编排：
| | 服务器编排 | 客户端编排 |
|---|---|---|
| 状态存储在 | 服务器会话存储 | 代理/调用者 |
| 步骤强制 | 服务器验证转换 | 代理负责 |
| 最适合 | 订单、支付、多阶段表单 | 状态轮询、幂等读取 |
| 会话头 | 需要`X-Session-ID` | 不需要 |
### 身份验证
在`AgentApp`上设置身份验证策略。客户端SDK根据清单声明的内容自动注入凭证。
```python
# API密钥（注入为X-API-Key头）
auth=AuthDef(type="api_key", header="X-API-Key")
# 承载令牌（注入为Authorization: Bearer ...）
auth=AuthDef(type="bearer")
# 不需要身份验证
auth=AuthDef(type="none")
```
在客户端：
```python
client = AgentClient(agent_id="my-agent", api_key="sk-...")
# 根据清单内容自动注入api_key
```
### 自定义会话存储
默认的`MemorySessionStore`适用于开发环境。在生产环境中，实现`SessionStore`协议，将会话存储在Redis、数据库或其他存储中。
```python
from agentgate.server.session import SessionStore, FlowSession
class RedisSessionStore:
async def create(self, flow_name: str, entry_step: str, allowed_next: list[str]) -> FlowSession: ...
async def get(self, session_id: str) -> FlowSession: ...
async def update(self, session: FlowSession) -> None: ...
agent_app = AgentApp(
title="MyShop",
session_store=RedisSessionStore(),
)
```
---
## HTTP协议参考
### 代理识别头
由客户端在每个请求上发送：
| 头部 | 值 | 必需 |
|---|---|---|
| `X-Agent-Request` | `true` | 是 |
| `X-Agent-ID` | 任何字符串标识符 | 否 |
| `X-Agent-Version` | `1.0` | 否 |
### 端点
| 方法 | 路径 | 描述 |
|---|---|---|
| `GET` | `/.well-known/agent-manifest.json` | 发现功能 |
| `POST` | `/agent/intents/{name}` | 执行意图 |
| `POST` | `/agent/flows/{name}/start` | 启动服务器编排的流程会话 |
| `POST` | `/agent/flows/{name}/step/{step_name}` | 继续服务器编排的步骤（需要`X-Session-ID`) |
| `GET` | `/agent/flows/{name}/session/{session_id}` | 检查服务器编排的会话 |
| `POST` | `/agent/flows/{name}/steps/{step_name}` | 执行客户端编排的步骤（无状态） |
### 流程启动响应
```json
{
"session_id": "550e8400-e29b-41d4-a716-446655440000",
"entry_step": "add_to_cart",
"allowed_next_steps": ["add_to_cart"],
"input_schema": { ... }
}
```
### 响应步骤
```json
{
"result": { ... },
"session_id": "550e8400-e29b-41d4-a716-446655440000",
"current_step": "add_to_cart",
"allowed_next_steps": ["place_order"],
"flow_complete": false
}
```
### 错误响应
所有错误都遵循标准格式：
```json
{
"error": "step_not_allowed",
"message": "Step 'place_order' cannot follow ''. Allowed next: ['add_to_cart']",
"current_step": "",
"allowed_next_steps": ["add_to_cart"]
}
```
| HTTP状态码 | 含义 |
|---|---|
| `400` | 缺少必需的头部（例如`X-Session-ID`) |
| `401` | 身份验证失败 |
| `404` | 意图、流程或会话未找到 |
| `422` | 无效的步骤转换或输入错误 |
---
## 错误处理 - 客户端
```python
from agentgate import (
ManifestNotFoundError, # 站点不支持agentgate
StepTransitionError, # 尝试跳过步骤 |
SessionNotFoundError, # 会话过期或无效 |
IntentError, # 意图执行失败 |
FlowNotFoundError, | 流程名称不在清单中 |
)
try:
site = await client.discover("https://example.com")
except ManifestNotFoundError:
print("该站点不支持agentgate")
try:
await session.step("place_order", ...)
except StepTransitionError as e:
print(f"Can't go there yet. Allowed: e.allowed")
```
连接信息
猜你喜欢
暂无推荐服务器