ftai-deep-agent 0.2.2

FtAi Agent Hub adapter for LangChain Deep Agent

ftai-deep-agent

将 LangChain DeepAgent 接入 FtAi Agent Hub 的适配器。

基于 ftai-langchain 构建，流式输出、工具调用、人机交互、自动重连，开箱即用。

如果你使用的是其他 LangChain / LangGraph 用法（非 DeepAgent），请直接使用 ftai-langchain。

安装

uv add ftai-deep-agent

快速开始

1. 配置环境变量

创建 .env 文件：

AGENT_SECRET=sk-ftai-ag-xxxxx
# AGENT_HUB_URL=wss://ftai.chat/api/v1/completion/external-agents/ws  # 可选，有默认值

AGENT_SECRET 由 FtAi Agent Hub 管理后台生成，格式以 sk-ftai-ag- 开头。

2. 编写 Agent

import asyncio
import json
import os

import httpx
from dotenv import load_dotenv
from deepagents import create_deep_agent
from ftai_deep_agent import AgentHubClient

load_dotenv()


async def get_ip_info(ip: str) -> str:
    """查询 IP 地址的地理位置信息。

    Args:
        ip: 要查询的 IP 地址，如 "8.8.8.8"。
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://ipinfo.io/{ip}/json")
        return json.dumps(resp.json(), ensure_ascii=False)


client = AgentHubClient(
    secret=os.environ["AGENT_SECRET"],
)

agent = create_deep_agent(
    model="claude-sonnet-4-6",
    tools=[get_ip_info],
    system_prompt="你是一个网络工具助手。",
)


async def main() -> None:
    await client.run(agent)


if __name__ == "__main__":
    asyncio.run(main())

uv run python main.py

工作流程

用户 ──> FtAi Agent Hub ──WebSocket──> AgentHubClient ──> DeepAgent (LLM)
                                         │
                                         ├── stream_text      (流式文本)
                                         ├── stream_thinking  (思考过程)
                                         ├── tool_call        (工具调用)
                                         └── message_end      (完成)

API 参考

AgentHubClient

from ftai_deep_agent import AgentHubClient

client = AgentHubClient(
    secret="sk-ftai-ag-...",        # Agent 密钥（必填）
    # agent_hub_url="wss://...",    # 可选，默认读取 AGENT_HUB_URL 环境变量或内置默认地址
    reconnect_initial=2.0,          # 重连初始间隔（秒）
    reconnect_max=60.0,             # 重连最大间隔（秒）
)

属性 / 方法	说明
client.agent_id	认证成功后的 Agent ID（只读）
client.human_in_loop_tool	人机交互工具，加入 tools 列表即可
await client.run(agent)	连接网关并处理请求（阻塞，自动重连）
await client.stop()	优雅关闭连接

重连策略

场景	行为
连接断开	指数退避重连（2s → 4s → 8s ... 最大 60s）
关闭码 4000（被新连接替换）	立即重连，不退避
认证成功	退避间隔重置为初始值
认证失败	抛出 AuthError，不重试

编写工具

工具就是带有 docstring 和 类型注解 的普通 Python 函数。

推荐使用 async def（SDK 运行在 async 事件循环中，同步 I/O 会阻塞整个 Agent）：

async def search_docs(query: str, top_k: int = 5) -> str:
    """从知识库中搜索相关文档。

    Args:
        query: 搜索关键词。
        top_k: 返回结果数量，默认 5。
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://your-api.example.com/search",
            params={"q": query, "limit": top_k},
        )
        return resp.text

示例

# 天气查询 Agent
uv run python examples/weather_agent/main.py

# IP 查询 Agent
uv run python examples/ip_info_agent/main.py