背景

iFlow CLI完成了第一版基于用户本地终端的人机交互模式，这东西是否只是一个个人开发助手？最近我们发现了新玩法，我们最近在和其他团队基于CLI搭建了业务的workflow，他们将自己业务能力通过SubAgent和MCP工具的方式进行了定义，把iFlow CLI集成到了自己的业务系统中，实现了很好的业务效果。

iFlow CLI：（One） Control Loop + （Many）Prompts +（Some）Tools，大道至简的Agent框架，可以释放模型的最大潜能。在很多业务上，可以充分发挥想象力，这种灵活的Agent架构发挥模型的最大潜能，给业务带来一些意想不到的价值。

我们基于业务需求，研发了一套包含iFlow CLI运行时的SDK开发工具。不同的业务基于不同语言的SDK可以将CLI嵌套到自己的业务应用当中。通信协议我们选择了目前CLI主流的ACP（Agent Client Protocol）协议，使用一个简单的命令就可以启动一个iFlow CLI的通过ACP协议传输的Server。

系统架构

iFlow CLI SDK通过ACP协议与本地或者远端的沙箱通过WebSocket双向通信，完成初始化、执行任务、输出结果等的信息交互。应用端可以直接集成iFlow CLI SDK来进行应用端的效果渲染和深度的任务执行规划。

效果

我们目前基于iFlow ACP Server已经无缝接入了Zed、AionUi、cherry-studio等项目

在Zed编辑器中使用iFlow CLI https://github.com/zed-industries/zed 65.8k ⭐️⭐️

AionUi中使用iFlow CLI https://github.com/iOfficeAI/AionUi 2k ⭐️⭐️

cherry-studio中使用iFlow CLI https://github.com/CherryHQ/cherry-studio 33.2k ⭐️⭐️

使用iFlow的ACP服务

下载安装iFlow CLI

mac/linux/ubuntu用户

# 一键安装脚本，会安装全部所需依赖
bash -c "$(curl -fsSL https://gitee.com/iflow-ai/iflow-cli/raw/main/install.sh)"

# 已有Node.js 22+
npm i -g @iflow-ai/iflow-cli@latest

已经安装过的话可以再次执行这个脚本，实现更新，后续CLI会自动更新

Windows用户

1.访问 https://nodejs.org/zh-cn/download 下载最新的 Node.js 安装程序

2.运行安装程序来安装 Node.js

3.重启终端：CMD 或 PowerShell

4.运行 npm install -g @iflow-ai/iflow-cli 来安装 iFlow CLI

5.运行 iflow 来启动 iFlow CLI

本地使用

# 启动本地stdio模式的acp服务
iflow --experimental-acp

# 启动WebSocket模式的acp服务
iflow --experimental-acp --port 端口号

远端使用

iflow --experimental-acp --port 端口号

iFlow CLI SDK（python版本）

iFlow CLI SDK 是一个用于与 iFlow CLI ACP Server进行编程交互的 Python SDK。它允许你通过 链接本地的iFlow CLI或者远端的iFlow CLI连接发送查询和接收响应，实现自动化的 AI 代理交互。

java版本的正在路上！

需要iFlow CLI >= 0.2.24版本

安装

pip install iflow-cli-sdk

快速开始

简单查询

最简单的使用方式是通过 query 函数发送单个查询：

from iflow_sdk import query
import asyncio

async def main():
    response = await query("什么是量子计算？")
    print(response)

asyncio.run(main())

交互式对话

对于更复杂的交互，可以使用 IFlowClient 类：

from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage
import asyncio

async def chat():
    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

asyncio.run(chat())

核心概念

IFlowClient

IFlowClient 是与 iFlow CLI 交互的主要接口。它管理 WebSocket 连接的生命周期，并提供发送和接收消息的方法。

from iflow_sdk import IFlowClient, IFlowOptions

options = IFlowOptions(
    url="ws://localhost:8090/acp",  # WebSocket URL
    auto_start_process=True  # 如果 iFlow 未运行则自动启动
)
async with IFlowClient(options) as client:
    # 使用 client
    pass

消息类型

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

AssistantMessage    AI 助手的文本响应。

from iflow_sdk import AssistantMessage

async for message in client.receive_messages():
    if isinstance(message, AssistantMessage):
        print(f"助手: {message.chunk.text}")
        if message.agent_id:
            print(f"来自代理: {message.agent_id}")

ToolCallMessage

工具调用消息，表示 AI 正在执行某个工具。

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

任务计划消息，展示 AI 的执行计划。

from iflow_sdk import PlanMessage

async for message in client.receive_messages():
    if isinstance(message, PlanMessage):
        for entry in message.entries:
            print(f"- [{entry.status}] {entry.content}")
            print(f"  优先级: {entry.priority}")

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

高级用法

自定义配置

后续会支持更多自定义配置，包括system prompt等

from iflow_sdk import IFlowClient, IFlowOptions

options = IFlowOptions(
    url="ws://localhost:8090/acp",  # WebSocket URL
    auto_start_process=True,  # 自动启动 iFlow
    timeout=30.0,  # 连接和操作超时时间（秒）
    log_level="INFO"  # 日志级别
)
client = IFlowClient(options)

手动生命周期管理

如果不使用上下文管理器，需要手动管理连接：

client = IFlowClient()
await client.connect()
try:
    await client.send_message("你好")
    async for message in client.receive_messages():
        # 处理消息
        pass
finally:
    await client.close()

错误处理

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}")

完整示例

交互式聊天机器人

#!/usr/bin/env python3
"""交互式 iFlow 聊天机器人"""

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def chatbot():
    print("iFlow 聊天机器人 (输入 'quit' 退出)")
    print("-" * 50)
    
    async with IFlowClient() as client:
        while True:
            # 获取用户输入
            try:
                user_input = input("\n你: ")
            except EOFError:
                break
                
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("再见！")
                break
            
            # 发送到 iFlow
            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())

代码分析工具

#!/usr/bin/env python3
"""使用 iFlow 分析代码文件"""

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"请分析以下代码并提供改进建议：\n\n{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]))

任务自动化

#!/usr/bin/env python3
"""使用 iFlow 自动化开发任务"""

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

async def automate_task(task_description):
    async with IFlowClient() as client:
        # 发送任务描述
        await client.send_message(task_description)
        
        plan_shown = False
        async for message in client.receive_messages():
            # 显示计划
            if isinstance(message, PlanMessage) and not plan_shown:
                print("执行计划：")
                for entry in message.entries:
                    status_icon = "✅" if entry.status == "completed" else "⏳"
                    print(f"{status_icon} {entry.content}")
                print("\n执行中...\n")
                plan_shown = True
            
            # 显示进度
            elif isinstance(message, AssistantMessage):
                print(message.chunk.text, end="", flush=True)
            
            # 任务完成
            elif isinstance(message, TaskFinishMessage):
                print("\n\n任务完成！")
                break

# 示例任务
task = """
创建一个简单的 Python 函数，用于：
1. 读取 CSV 文件
2. 计算数值列的平均值
3. 将结果保存到新文件
"""

if __name__ == "__main__":
    asyncio.run(automate_task(task))

故障排除

iFlow 未启动

如果遇到连接错误，确保 iFlow CLI 正在运行：

# 检查 iFlow 是否运行
ps aux | grep iflow

# 手动启动 iFlow
iflow

端口冲突

如果默认端口 8090 被占用，可以指定其他端口：

from iflow_sdk import IFlowClient, IFlowOptions

options = IFlowOptions(url="ws://localhost:8080/acp")
client = IFlowClient(options)

超时问题

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

from iflow_sdk import IFlowClient, IFlowOptions

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

业务合作

基于iFlow CLI的ACP协议可以做很多事情，iFlow CLI SDK是基于ACP协议开发的，你也可以直接使用iFlow CLI原生的服务来直接serving你的应用，例如Zed编辑器和AionUI。这套流程已经在很多业务上进行了尝试和落地。如果你还在对Agent执行框架的选择有所纠结，现在可以不用等了，直接选择这套就完事了，足够轻量！很期待有更多的业务来和我们合作，一起推动iFlow CLI和业务上的发展。