AI Agent Gateway

ai代理网关

将AI代理部署为生产服务。

其他框架帮助您定义代理的功能。 这处理了它周围的一切——HTTP服务器、会话管理、SSE流、工具调度、人在环审批和代码执行沙盒。

从系统提示开始。根据需要添加MCP工具、本地Python工具、技能和代码执行。需要Python>=3.10。

安装+快速启动

create_agent() 是通往工作代理服务器的最快路径。默认情况下，它使用Anthropic，您可以使用以下命令切换到OpenAI provider="openai".使用 create_gateway_app() 当您需要较低级别的运行时控制时。

安装包装并 uvicorn:

pip install "ai-agent-gateway[anthropic]" uvicorn
export ANTHROPIC_API_KEY="your-anthropic-api-key"

对于OpenAI来说：

pip install "ai-agent-gateway[openai]" uvicorn
export OPENAI_API_KEY="your-openai-api-key"

创建 agent.py:

from agent_gateway import create_agent

app = create_agent("You are a concise research assistant.")

运行服务器：

uvicorn agent:app --reload --port 8000

创建会话令牌：

SESSION_TOKEN=$(curl -s http://127.0.0.1:8000/api/chat/init \
  -H 'Content-Type: application/json' \
  -d '{"api_key":"local-demo-key"}' \
  | python3 -c 'import json,sys; print(json.load(sys.stdin)["session_token"])')

与客服聊天：

curl -N http://127.0.0.1:8000/api/chat \
  -H "Authorization: Bearer $SESSION_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"messages": [{"role": "user", "content": "Give me three bullet points on why SSE is useful for chat UIs."}]}'

您将获得一个SSE流，如下所示：

data: {"type":"text_delta","text":"- SSE lets the server push tokens as they are generated.\n"}

data: {"type":"text_delta","text":"- The browser can render partial output without polling.\n"}

data: {"type":"stream_complete","usage":{"input_tokens":...,"output_tokens":...}}

完整的5分钟演练： 快速启动

特性

• FastAPI服务器工厂 /api/chat/init, /api/chat, /api/chat/tool-result, /api/chat/tool-approval，以及 /api/health
• 文本增量、思维增量、工具调用、批准请求、工具输出块、重试和完成的SSE事件流
• 具有范围批准和隔离代码执行目录的JWT会话
• 从内联配置或 ~/.claude.json
• 与MCP工具具有相同调度循环的本地Python工具处理程序
• 首选Docker和子进程回退执行代码
• Markdown技能文件（每个任务的提示+配置）和子代理通过内置 run_agent 工具
• Anthropic和OpenAI提供商通过 create_agent() 或 create_gateway_app()

您带来了系统提示、工具（MCP服务器、本地Python处理程序或两者）和运行时策略。网关处理其他所有事务。

渐进式示例

第1层：仅系统提示

from agent_gateway import create_agent

app = create_agent("You are a helpful assistant for spreadsheet users.")

第2层：添加MCP工具

这使用内联MCP服务器配置。下面的示例假设安装了Node.js，因为它运行一个 npx-基于MCP服务器。

from agent_gateway import create_agent

app = create_agent(
  "You can inspect and edit files when needed.",
  mcp_servers={
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "."],
    }
  },
)

第3层：添加本地工具

from agent_gateway import create_agent

async def summarize_csv(tool_input, **_kwargs):
  path = tool_input["path"]
  return {"summary": f"Would summarize {path}"}, None

app = create_agent(
  "Use the summarize_csv tool when the user asks for a file summary.",
  tool_handlers={"summarize_csv": summarize_csv},
  tool_definitions=[
    {
      "name": "summarize_csv",
      "description": "Summarize a CSV file on disk.",
      "input_schema": {
        "type": "object",
        "properties": {
          "path": {"type": "string", "description": "Path to the CSV file."}
        },
        "required": ["path"],
      },
    }
  ],
)

第4层：添加代码执行和技能

code_execution=True 如果可用，则更喜欢Docker，否则则退回到本地子流程执行。

from agent_gateway import create_agent

app = create_agent(
  "Use code execution for calculations and run_agent for focused subtasks.",
  code_execution=True,       # Adds code_execute tool (Docker preferred, subprocess fallback)
  skills_dir="skills",       # Each .md file becomes a named skill for run_agent
)

毕业生：切换到 create_gateway_app()

使用 create_gateway_app() 当您需要自定义审批逻辑、通道感知运行时、拦截器、多个运行时配置文件或更深层次的生产挂钩时。

from agent_gateway import (
  AnthropicProvider, ChatRuntime, GatewayServerConfig, create_gateway_app,
)

# Full control: custom providers, approval logic, channel routing, interceptors.
# See examples/07-full-production/ for the complete version.
app = create_gateway_app(
  GatewayServerConfig(
    build_chat_runtime=my_runtime_factory,
    default_provider=AnthropicProvider(),
  )
)

这些示例的可运行版本存在于 examples/.

建筑

Client
  |
  |-- POST /api/chat/init --> JWT session token
  |
  |-- POST /api/chat (Bearer token, SSE stream)
  |       |
  |       v
  |   ChatRuntime (built per-request)
  |       |
  |       v
  |   AgentRunner (model loop: stream -> tool calls -> dispatch -> resume)
  |       |
  |       v
  |   ToolDispatcher
  |       |-- interceptors (rate limits, custom policies)
  |       |-- approval check (session-scoped)
  |       |-- local Python handler
  |       |-- MCP server (stdio)
  |       |-- code_execute (Docker / subprocess)
  |       |-- run_agent (sub-agent with own runner)
  |       |
  |       v
  |   EventLog --> SSE events to client
  |
  |-- POST /api/chat/tool-approval (human-in-the-loop)

同一后端可以服务于多个前端。通过 context.channel 在不重写代理循环的情况下，塑造每个客户端的运行时行为。

比较

何时使用此包：您希望用户或客户端通过HTTP与代理通信，而不希望自己构建会话、SSE、审批和工具服务基础设施。

何时不使用此包：您需要显式的图编排，或者您正在构建一个不需要可重用服务器运行时的一次性笔记本或脚本。

你也可以把它们结合起来。例如，LangGraph工作流可以位于 ai-agent-gateway HTTP表面。

文档

• 快速启动
• HTTP API
• 建筑
• 比较
• api参考
• 贡献

许可证

麻省理工学院