ftai-deep-agent 0.1.0

SDK for integrating LangChain Deep Agent into downstream projects

ftai-deep-agent

将 LangChain Deep Agent 接入 Agent Gateway 的 Python SDK。

通过 WebSocket 长连接，让你的 AI Agent 无缝对接平台能力 -- 流式输出、工具调用、人机交互、自动重连，开箱即用。

环境要求

• Python >= 3.12
• uv (包管理器)

快速开始

1. 安装依赖

# 克隆项目后
uv sync

如需在你的项目中使用此 SDK：

uv add ftai-deep-agent

2. 配置环境变量

创建 .env 文件：

AGENT_GATEWAY_URL=wss://ftai.chat/api/v1/agent-gateway/ws
AGENT_SECRET=sk-ftai-ag-xxxxx

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

3. 编写你的第一个 Agent

# main.py
import asyncio
import json
import os

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

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 = AgentGatewayClient(
    gateway_url=os.environ["AGENT_GATEWAY_URL"],
    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

启动后 Agent 会自动连接网关、等待用户消息、流式返回回复。就这么简单。

核心概念

工作流程

用户 ──> Agent Gateway ──WebSocket──> ftai-deep-agent SDK ──> DeepAgent (LLM)
                                           │
                                           ├── stream_text      (流式文本)
                                           ├── stream_thinking  (思考过程)
                                           ├── tool_call        (工具调用)
                                           └── message_end      (完成)

SDK 负责中间这层桥接 -- 你只需要关注 Agent 的业务逻辑和工具定义。

三步构建 Agent

步骤	做什么	用什么
1. 定义工具	编写普通 Python 函数，添加 docstring	函数 + 类型注解
2. 创建 Agent	选择模型，组装工具和 system prompt	create_deep_agent()
3. 连接网关	创建客户端并运行	AgentGatewayClient

编写工具

工具就是带有 docstring 和 类型注解 的普通 Python 函数。DeepAgent 会自动将它们转换为 LLM 可调用的工具。

通常你可以在这里调用你使用 Rust 写的 API 接口。

推荐使用 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

纯计算型工具可以用普通 def，框架会自动放到线程池运行：

def calculate_bmi(height_m: float, weight_kg: float) -> str:
    """计算 BMI 指数。

    Args:
        height_m: 身高（米）。
        weight_kg: 体重（公斤）。
    """
    bmi = weight_kg / (height_m ** 2)
    return json.dumps({"bmi": round(bmi, 1)})

最佳实践：

• I/O 操作用 async def — 网络请求、数据库查询等，使用 httpx.AsyncClient 而非 requests
• 纯计算用 def 即可，框架自动处理线程
• 返回 str 类型（JSON 字符串最佳，LLM 解析友好）
• docstring 写清楚功能和参数含义 -- LLM 靠它理解何时使用
• 参数用基本类型（str、int、float、bool），加上合理的默认值
• 函数名使用 snake_case，语义清晰

API 参考

AgentGatewayClient

from ftai_deep_agent import AgentGatewayClient

client = AgentGatewayClient(
    gateway_url: str,       # WebSocket 地址 (wss://...)
    secret: str,            # Agent 密钥 (sk-ftai-ag-...)
    reconnect_initial: float = 2.0,   # 重连初始间隔（秒）
    reconnect_max: float = 60.0,      # 重连最大间隔（秒）
)

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

AuthError

from ftai_deep_agent import AuthError

认证失败时抛出，不会自动重试。请检查 AGENT_SECRET 是否正确。

重连策略

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

完整示例：天气查询 Agent

项目中 examples/weather_agent/ 提供了完整的天气查询 Agent 示例：

examples/weather_agent/
├── main.py       # 入口：创建客户端并连接网关
├── agent.py      # Agent 定义：模型 + 工具 + system prompt
├── tools.py      # 工具函数：城市搜索、实时天气、天气预报、生活指数
└── qweather.py   # 和风天气 API 客户端封装

运行方式：

# 补充 .env 中的天气 API 配置
QWEATHER_API_HOST=https://...
QWEATHER_API_KEY=your_key
OPENROUTER_API_KEY=your_key    # 示例使用 OpenRouter 模型

uv run python examples/weather_agent/main.py

项目结构

src/ftai_deep_agent/
├── __init__.py      # 公开 API：AgentGatewayClient, AuthError
├── gateway.py       # WebSocket 客户端核心实现
├── protocol.py      # 协议常量、JSON 编解码、消息构建器
└── _convert.py      # OpenAI <-> LangChain 消息格式转换

开发

# 安装全部依赖
uv sync

# 添加新依赖
uv add <package>

# 运行示例
uv run python examples/weather_agent/main.py

License

Private