Metadata-Version: 2.4
Name: ftai-deep-agent
Version: 0.1.0
Summary: SDK for integrating LangChain Deep Agent into downstream projects
Requires-Python: >=3.12
Requires-Dist: cryptography>=46.0.6
Requires-Dist: deepagents>=0.4.12
Requires-Dist: httpx[socks]>=0.28.1
Requires-Dist: pyjwt>=2.12.1
Requires-Dist: python-dotenv>=1.2.2
Requires-Dist: tavily-python>=0.7.23
Requires-Dist: websockets>=13.0
Description-Content-Type: text/markdown

# ftai-deep-agent

将 [LangChain Deep Agent](https://docs.langchain.com/) 接入 Agent Gateway 的 Python SDK。

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

## 环境要求

- Python >= 3.12
- [uv](https://docs.astral.sh/uv/) (包管理器)

## 快速开始

### 1. 安装依赖

```bash
# 克隆项目后
uv sync
```

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

```bash
uv add ftai-deep-agent
```

### 2. 配置环境变量

创建 `.env` 文件：

```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

```python
# 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())
```

```bash
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）：

```python
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`，框架会自动放到线程池运行：

```python
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`

```python
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`

```python
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 客户端封装
```

运行方式：

```bash
# 补充 .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 消息格式转换
```

## 开发

```bash
# 安装全部依赖
uv sync

# 添加新依赖
uv add <package>

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

## License

Private
