SDK

概述

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, AssistantMessage, TaskFinishMessage

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():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="\n", flush=True)
            elif isinstance(message, TaskFinishMessage):
                break

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("你的问题")
    async for message in client.receive_messages():
        # 处理消息
        pass

# 使用自定义配置
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("你的问题")
    async for message in client.receive_messages():
        # 处理消息
        pass

消息类型

SDK 支持多种消息类型，对应 iFlow 协议的不同响应：

AssistantMessage - AI 助手响应

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def handle_assistant_message():
    async with IFlowClient() as client:
        await client.send_message("请介绍一下Python")
        
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="\n", flush=True)
            elif isinstance(message, TaskFinishMessage):
                break

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

ToolCallMessage - 工具调用

import asyncio
from iflow_sdk import IFlowClient, ToolCallMessage, ToolCallStatus, TaskFinishMessage

async def handle_tool_calls():
    async with IFlowClient() as client:
        # 注意：这个示例演示如何查看工具调用消息
        # iFlow 会要求提供完整路径和内容来创建文件
        await client.send_message("列出当前目录的文件")
        
        async for message in client.receive_messages():
            if isinstance(message, ToolCallMessage):
                print(f"工具: {message.label}")  
            elif isinstance(message, TaskFinishMessage):
                break

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

PlanMessage - 任务计划

import asyncio
from iflow_sdk import IFlowClient, PlanMessage, TaskFinishMessage

async def show_plan():
    async with IFlowClient() as client:
        await client.send_message("帮我创建一个Python项目结构")
        
        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}")
            elif isinstance(message, TaskFinishMessage):
                break

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

TaskFinishMessage - 任务完成

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage, StopReason

async def check_completion():
    async with IFlowClient() as client:
        await client.send_message("计算 1+1")
        
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="", flush=True)
            elif isinstance(message, TaskFinishMessage):
                print()  # 换行
                if message.stop_reason == StopReason.END_TURN:
                    print("任务正常完成")
                elif message.stop_reason == StopReason.MAX_TOKENS:
                    print("达到最大令牌限制")
                break  # TaskFinishMessage 表示对话结束

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

常见用例

交互式聊天机器人

#!/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 = "解释一下什么是升华现象"
    
    # query_stream 返回文本块的异步生成器
    async for chunk in query_stream(prompt):
        print(chunk, end="", flush=True)
    print()  # 最后换行

asyncio.run(stream_example())

高级配置

手动进程管理

如果需要手动管理 iFlow 进程：

import asyncio
from iflow_sdk import IFlowClient, IFlowOptions, AssistantMessage, TaskFinishMessage

async def manual_process_example():
    # 禁用自动进程管理
    options = IFlowOptions(
        auto_start_process=False,
        url="ws://localhost:8090/acp"  # 连接到已存在的 iFlow
    )
    
    async with IFlowClient(options) as client:
        await client.send_message("你的问题")
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="", flush=True)
            elif isinstance(message, TaskFinishMessage):
                break

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

注意手动模式需要您单独启动 iFlow：
iflow --experimental-acp --port 8090

错误处理

SDK 提供了详细的错误处理机制：

import asyncio
from iflow_sdk import IFlowClient, ConnectionError, TimeoutError, AssistantMessage, TaskFinishMessage

async def error_handling_example():
    try:
        async with IFlowClient() as client:
            await client.send_message("测试")
            async for message in client.receive_messages():
                if isinstance(message, AssistantMessage):
                    print(message.chunk.text, end="", flush=True)
                elif isinstance(message, TaskFinishMessage):
                    break
    except ConnectionError as e:
        print(f"连接错误: {e}")
    except TimeoutError as e:
        print(f"超时错误: {e}")
    except Exception as e:
        print(f"未知错误: {e}")

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

同步调用

对于需要同步调用的场景：

from iflow_sdk import query_sync, IFlowOptions

# 同步调用，带超时控制
options = IFlowOptions(timeout=30.0)
response = query_sync("你的问题", options=options)
print(response)

API 参考

核心类

类名	描述
`IFlowClient`	主客户端类，管理与 iFlow 的连接
`IFlowOptions`	配置选项类
`RawDataClient`	访问原始协议数据的客户端

消息类型

消息类型	描述	主要属性
`AssistantMessage`	AI 助手的文本响应	`chunk.text`, `agent_id`
`ToolCallMessage`	工具执行请求和状态	`label`, `status`, `content`, `icon`
`PlanMessage`	结构化任务计划	`entries` (包含 `content`, `priority`, `status`)
`TaskFinishMessage`	任务完成信号	`stop_reason`

故障排除

连接问题

如果遇到连接错误，请检查：

iFlow 是否已安装
```
iflow --version
```

端口是否被占用

# Linux/Mac
lsof -i :8090

# Windows
netstat -an | findstr :8090

手动测试连接
```
iflow --experimental-acp --port 8090
```

超时问题

对于长时间运行的任务，可以增加超时时间：

from iflow_sdk import IFlowClient, IFlowOptions

options = IFlowOptions(timeout=600.0)  # 10分钟超时
client = IFlowClient(options)

日志调试

启用详细日志以便调试：

import logging
from iflow_sdk import IFlowClient, IFlowOptions

# 设置日志级别
logging.basicConfig(level=logging.DEBUG)

options = IFlowOptions(log_level="DEBUG")
client = IFlowClient(options)

概述​

系统要求​

安装​

快速开始​

基础示例​

简单查询​

核心概念​

IFlowClient​

消息类型​

AssistantMessage - AI 助手响应​

ToolCallMessage - 工具调用​

PlanMessage - 任务计划​

TaskFinishMessage - 任务完成​

常见用例​

交互式聊天机器人​

流式响应处理​

高级配置​

手动进程管理​

错误处理​

同步调用​

API 参考​

核心类​

消息类型​

故障排除​

连接问题​

超时问题​

日志调试​

概述

系统要求

安装

快速开始

基础示例

简单查询

核心概念

IFlowClient

消息类型

AssistantMessage - AI 助手响应

ToolCallMessage - 工具调用

PlanMessage - 任务计划

TaskFinishMessage - 任务完成

常见用例

交互式聊天机器人

流式响应处理

高级配置

手动进程管理

错误处理

同步调用

API 参考

核心类

消息类型

故障排除

连接问题

超时问题

日志调试