概述
iFlow CLI SDK 是一个用于与 iFlow CLI 进行编程交互的 SDK。它通过代理通信协议（ACP）允许开发者构建具有对话、工具执行和任务规划能力的 AI 驱动应用程序。
核心特性：SDK 自动管理 iFlow 进程 - 无需手动配置！
目前提供 Python SDK，Java 版本正在路上，敬请期待！
系统要求
-
Python: 3.8 或更高版本
-
iFlow CLI: 0.2.24 或更高版本
-
操作系统: Windows、macOS、Linux
安装
pip install iflow-cli-sdk
快速开始
基础示例
SDK 会自动检测并启动 iFlow 进程，无需手动配置：
import asyncio
from iflow_sdk import IFlowClient
async def main():
# SDK 自动处理：
# 1. 检测 iFlow 是否已安装
# 2. 启动 iFlow 进程（如果未运行）
# 3. 查找可用端口并建立连接
# 4. 退出时自动清理资源
async with IFlowClient() as client:
await client.send_message("你好，iFlow！")
async for message in client.receive_messages():
print(message)
asyncio.run(main())
简单查询
最简单的使用方式是通过 query 函数：
from iflow_sdk import query
import asyncio
async def main():
response = await query("法国的首都是哪里？")
print(response) # 输出：法国的首都是巴黎。
asyncio.run(main())
核心概念
IFlowClient
IFlowClient 是与 iFlow CLI 交互的主要接口，管理 WebSocket 连接的生命周期：
from iflow_sdk import IFlowClient, IFlowOptions
# 使用默认配置（自动管理进程）
async with IFlowClient() as client:
await client.send_message("你的问题")
# 使用自定义配置
options = IFlowOptions(
url="ws://localhost:8090/acp", # WebSocket URL
auto_start_process=True, # 自动启动 iFlow
timeout=30.0 # 超时时间（秒）
)
async with IFlowClient(options) as client:
await client.send_message("你的问题")
消息类型
SDK 支持多种消息类型，对应 iFlow 协议的不同响应：
AssistantMessage - AI 助手响应
from iflow_sdk import AssistantMessage
async for message in client.receive_messages():
if isinstance(message, AssistantMessage):
print(message.chunk.text, end="", flush=True)
if message.agent_id:
print(f"\\n来自代理: {message.agent_id}")
ToolCallMessage - 工具调用
from iflow_sdk import ToolCallMessage, ToolCallStatus
async for message in client.receive_messages():
if isinstance(message, ToolCallMessage):
print(f"工具: {message.tool_name}")
print(f"状态: {message.status.value}")
if message.status == ToolCallStatus.COMPLETED:
print(f"结果: {message.result}")
PlanMessage - 任务计划
from iflow_sdk import PlanMessage
async for message in client.receive_messages():
if isinstance(message, PlanMessage):
print("执行计划：")
for entry in message.entries:
status_icon = "✅" if entry.status == "completed" else "⏳"
print(f"{status_icon} [{entry.priority}] {entry.content}")
TaskFinishMessage - 任务完成
from iflow_sdk import TaskFinishMessage, StopReason
async for message in client.receive_messages():
if isinstance(message, TaskFinishMessage):
if message.stop_reason == StopReason.STOP_SEQUENCE:
print("任务正常完成")
elif message.stop_reason == StopReason.MAX_TOKENS:
print("达到最大令牌限制")
break
常见用例
交互式聊天机器人
#!/usr/bin/env python3
import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage
async def chatbot():
print("iFlow 聊天机器人 (输入 'quit' 退出)")
print("-" * 50)
async with IFlowClient() as client:
while True:
user_input = input("\\n你: ")
if user_input.lower() in ['quit', 'exit', 'q']:
print("再见！")
break
await client.send_message(user_input)
print("iFlow: ", end="", flush=True)
async for message in client.receive_messages():
if isinstance(message, AssistantMessage):
print(message.chunk.text, end="", flush=True)
elif isinstance(message, TaskFinishMessage):
print() # 换行
break
if __name__ == "__main__":
asyncio.run(chatbot())
流式响应处理
from iflow_sdk import query_stream
import asyncio
async def stream_example():
prompt = "解释量子计算的基本原理"
async for chunk in query_stream(prompt):
print(chunk, end="", flush=True)
print() # 最后换行
asyncio.run(stream_example())
代码分析工具
#!/usr/bin/env python3
import asyncio
import sys
from pathlib import Path
from iflow_sdk import IFlowClient, AssistantMessage, ToolCallMessage, TaskFinishMessage
async def analyze_code(file_path):
code_content = Path(file_path).read_text()
prompt = f\"\"\"请分析以下代码并提供改进建议：
{code_content}
\"\"\"
async with IFlowClient() as client:
await client.send_message(prompt)
print("分析中...\\n")
async for message in client.receive_messages():
if isinstance(message, AssistantMessage):
print(message.chunk.text, end="", flush=True)
elif isinstance(message, ToolCallMessage):
if message.tool_name:
print(f"\\n[使用工具: {message.tool_name}]")
elif isinstance(message, TaskFinishMessage):
break
if __name__ == "__main__":
if len(sys.argv) != 2:
print("用法: python analyze_code.py <文件路径>")
sys.exit(1)
asyncio.run(analyze_code(sys.argv[1]))
高级配置
手动进程管理
如果需要手动管理 iFlow 进程：
from iflow_sdk import IFlowClient, IFlowOptions
# 禁用自动进程管理
options = IFlowOptions(
auto_start_process=False,
url="ws://localhost:8090/acp" # 连接到已存在的 iFlow
)
async with IFlowClient(options) as client:
await client.send_message("你的问题")
注意 手动模式需要您单独启动 iFlow：
iflow --experimental-acp --port 8090
错误处理
SDK 提供了详细的错误处理机制：
from iflow_sdk import IFlowClient, ConnectionError, TimeoutError
try:
async with IFlowClient() as client:
await client.send_message("测试")
async for message in client.receive_messages():
# 处理消息
pass
except ConnectionError as e:
print(f"连接错误: {e}")
except TimeoutError as e:
print(f"超时错误: {e}")
except Exception as e:
print(f"未知错误: {e}")
同步调用
对于需要同步调用的场景：
from iflow_sdk import query_sync
# 同步调用，带超时控制
response = query_sync("你的问题", timeout=30.0)
print(response)
API 参考
核心类
| 类名 | 描述 |
|---|---|
IFlowClient |
主客户端类，管理与 iFlow 的连接 |
IFlowOptions |
配置选项类 |
RawDataClient |
访问原始协议数据的客户端 |
消息类型
| 消息类型 | 描述 |
|---|---|
AssistantMessage |
AI 助手的文本响应 |
ToolCallMessage |
工具执行请求和状态 |
PlanMessage |
结构化任务计划 |
TaskFinishMessage |
任务完成信号 |
工具函数
| 函数 | 描述 |
|---|---|
query(prompt) |
发送查询并返回完整响应 |
query_stream(prompt) |
返回流式响应的异步生成器 |
query_sync(prompt) |
同步查询（带超时） |
故障排除
连接问题
如果遇到连接错误，请检查：
-
iFlow 是否已安装
iflow --version -
端口是否被占用
# Linux/Mac lsof -i :8090 # Windows netstat -an | findstr :8090 -
手动测试连接
iflow --experimental-acp --port 8090
超时问题
对于长时间运行的任务，可以增加超时时间：
options = IFlowOptions(timeout=600.0) # 10分钟超时
client = IFlowClient(options)
日志调试
启用详细日志以便调试：
import logging
# 设置日志级别
logging.basicConfig(level=logging.DEBUG)
options = IFlowOptions(log_level="DEBUG")
client = IFlowClient(options)