---
title: "AG2"
description: "将 AG2 多Agent 系统连接 TwexAPI，实现推文搜索、资料查询、时间线与委派研究工作流。"
---

[AG2](https://github.com/ag2ai/ag2) 是开源 Python 多AI Agent框架。TwexAPI 尚未提供原生 AG2 搜索工具包。通过 `MCPToolkit` 将 AG2 AI Agent连接到 [TwexAPI MCP 服务器](/mcp/overview)，使AI Agent可调用 `explore` 与 `twexapi_request`，同时将 API 凭证保留在基础设施内。

保留 API 返回的每条 tweet ID、用户 ID 与游标。

## 为何将 AG2 与 TwexAPI 搭配使用？

AG2 对工具暴露、委派与中间件提供显式控制。与 TwexAPI MCP 配对时，可在不硬编码每条 REST 路由的情况下做多AI Agent研究。

| 边界 | AG2 控制 | 收益 |
| ----------------- | ------------------------- | ---------------------------------------------------------------------- |
| 工具构建 | `MCPToolkit(...)` | 限制为 `explore` 与 `twexapi_request`，或进一步过滤 |
| 模型输入 | MCP 工具 schema | 模型仅从已发现端点选路由 |
| 运行时作用域 | `Variable` | 执行时解析每用户或每租户 header |
| 委派 | `Agent.as_tool()` | 协调者上下文不包含搜索者的工具调用历史 |
| 传输 | `MCPServerConfig` | 指向 `https://api.twexapi.io/mcp` 并带 `x-api-key` |

适合研究、监控与报告AI Agent。无需模型决策的确定性作业请用 [Python SDK](/sdks/python) 或 [Prefect 集合](/guides/prefect)。

## 前置条件

* Python 3.10 或更高版本
* [TwexAPI API 密钥](https://twexapi.io/dashboard)
* AG2 支持的 LLM 提供商密钥

公开 X 读取无需 X Developer 凭证，使用 TwexAPI 认证即可。

## 安装 AG2

```bash
python -m pip install "ag2>=1.0.0" python-dotenv
```

同时安装模型提供商扩展。

```bash
python -m pip install "ag2[anthropic]>=1.0.0"
```

将密钥存放在源码控制之外。

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"
```

## 用 MCPToolkit 连接 TwexAPI MCP

凭证须留在基础设施内时使用客户端 `MCPToolkit`。TwexAPI MCP 暴露 `explore` 与 `twexapi_request`。

```python
import asyncio
import os

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.tools import MCPToolkit, MCPServerConfig
from dotenv import load_dotenv

load_dotenv()
config = AnthropicConfig(model="claude-sonnet-4-20250514")

twexapi_mcp = MCPToolkit(
    MCPServerConfig(
        server_url="https://api.twexapi.io/mcp",
        server_label="twexapi",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
        allowed_tools=["explore", "twexapi_request"],
    )
)

agent = Agent(
    "x-researcher",
    prompt=(
        "Search X for evidence before answering. "
        "Always call explore before twexapi_request. "
        "Quote tweet text verbatim and keep every tweet ID you receive. "
        "Ask for confirmation before read_only: false actions."
    ),
    config=config,
    tools=[twexapi_mcp],
)


async def main() -> None:
    reply = await agent.ask(
        "What are developers saying about AI agents on X this week? "
        "Return tweet IDs, authors, and a 5-bullet summary."
    )
    print(reply.body)


asyncio.run(main())
```

`headers` 中的 `x-api-key` 为必填。未认证请求 `https://api.twexapi.io/mcp` 返回 `401`。

## 按AI Agent角色限制工具

仅发现型AI Agent访问 `explore`；执行型AI Agent两者皆可。

```python
catalog_agent_tools = [
    MCPToolkit(
        MCPServerConfig(
            server_url="https://api.twexapi.io/mcp",
            headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
            allowed_tools=["explore"],
            server_label="twexapi-catalog",
        )
    )
]

execution_agent_tools = [
    MCPToolkit(
        MCPServerConfig(
            server_url="https://api.twexapi.io/mcp",
            headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
            allowed_tools=["explore", "twexapi_request"],
            server_label="twexapi-execute",
        )
    )
]
```

## 在多AI Agent团队中委派搜索

`Agent.as_tool()` 将AI Agent暴露为另一AI Agent 的工具。协调者收到委派者的最终答案，而非其内部工具调用历史。

```python
searcher = Agent(
    "searcher",
    prompt=(
        "Use TwexAPI MCP to search public X posts. "
        "Call explore first. Return tweet text with IDs. Do not summarise away IDs."
    ),
    config=config,
    tools=[twexapi_mcp],
)

analyst = Agent(
    "analyst",
    prompt="Turn tweet records into a factual brief. Keep every tweet ID.",
    config=config,
)

coordinator = Agent(
    "coordinator",
    prompt="Delegate the search, then pass the tweets to the analyst.",
    config=config,
    tools=[
        searcher.as_tool(description="Search public X posts and return raw tweet records."),
        analyst.as_tool(description="Analyse tweet records. Pass them in the context parameter."),
    ],
)

reply = await coordinator.ask("Brief me on this week's discussion of AI agents on X.")
print(reply.body)
```

## Handoff 清单

持久化 MCP 响应中的 durable 字段，使后续工作流步骤不依赖聊天历史。

| 数据类型 | 存储 |
| --- | --- |
| 推文 | `tweet_id`、`text`、`author_username`、`created_at`、`has_more`、`next_cursor`、原始查询 |
| 资料 | `user_id`、`username`、`name`、`description`、`followers_count`、源查找 |
| 趋势 | 国家、主题、内容标签、推文行、请求筛选 |
| 写操作 | `tweet_id`、路由名、状态、确认记录 |

完整清单见 [Agent MCP Handoff](/mcp/agent-handoff)。

## 分页

`explore` 返回分页路由时，通过 `twexapi_request` 原样传回文档化游标字段。游标视为 opaque 字符串。按 `tweet_id` 或 `user_id` 去重。

```python
checkpoint = {
    "route_used": "/twitter/advanced_search/page",
    "query": "AI agents",
    "has_more": True,
    "next_cursor": "cursor_123",
}
```

## 与 Docs MCP 组合

Agent 应先搜索 TwexAPI 文档再选路由时，添加 [Docs MCP 服务器](/mcp/docs-mcp)。

```python
tools=[
    MCPToolkit(
        MCPServerConfig(
            server_url="https://docs.twexapi.io/mcp",
            server_label="twexapi-docs",
        )
    ),
    twexapi_mcp,
]
```

## 处理失败

MCP 与 REST 错误通过工具包以 HTTP 失败形式呈现。重试前按状态分支。

| 状态 | 操作 |
| ------ | ------ |
| `400` | 修正请求。切勿原样重试。 |
| `401` | 检查 `x-api-key` header 或 Bearer token。 |
| `403` | 检查账号访问、额度与写权限。 |
| `429` | 退避并保留游标。 |
| `5xx` | 有界退避重试。 |

需要重试、审批门或审计日志时，用 AG2 工具中间件包装 `MCPToolkit`。

## 提供商侧 MCP（仅 Anthropic）

若面向 Anthropic 且接受将凭证转发给提供商，可用 `MCPServerTool` 替代 `MCPToolkit`。需跨提供商部署且 API 密钥须留在基础设施内时，优先 `MCPToolkit`。

## 相关指南

* [AG2 MCP Servers documentation](https://docs.ag2.ai/docs/user-guide/tools/mcp_servers/)
* [MCP Server](/mcp/overview)
* [MCP Tools](/mcp/tools)
* [LangChain](/guides/langchain)
* [CrewAI](/guides/crewai)
* [Python SDK](/sdks/python)