SDK

Overview

The iFlow CLI SDK is a software development kit for programmatic interaction with iFlow CLI. Through the Agent Communication Protocol (ACP), it enables developers to build AI-powered applications with conversation, tool execution, and task planning capabilities.

✨ Core Feature: SDK automatically manages iFlow processes - no manual configuration needed!

Currently provides Python SDK, with Java version coming soon!

System Requirements

Python: 3.8 or higher
iFlow CLI: 0.2.24 or higher
Operating System: Windows, macOS, Linux

Installation

pip install iflow-cli-sdk

Quick Start

Basic Example

The SDK automatically detects and starts iFlow processes without manual configuration:

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def main():
    # SDK automatically handles:
    # 1. Detect if iFlow is installed
    # 2. Start iFlow process (if not running)
    # 3. Find available port and establish connection
    # 4. Cleanup resources on exit
    async with IFlowClient() as client:
        await client.send_message("Hello, 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())

Simple Query

The simplest usage is through the query function:

from iflow_sdk import query
import asyncio

async def main():
    response = await query("What is the capital of France?")
    print(response)  # Output: The capital of France is Paris.

asyncio.run(main())

Core Concepts

IFlowClient

IFlowClient is the main interface for interacting with iFlow CLI, managing WebSocket connection lifecycle:

from iflow_sdk import IFlowClient, IFlowOptions

# Use default configuration (automatic process management)
async with IFlowClient() as client:
    await client.send_message("Your question")
    async for message in client.receive_messages():
        # Process messages
        pass

# Use custom configuration
options = IFlowOptions(
    url="ws://localhost:8090/acp",  # WebSocket URL
    auto_start_process=True,         # Auto-start iFlow
    timeout=30.0                     # Timeout (seconds)
)
async with IFlowClient(options) as client:
    await client.send_message("Your question")
    async for message in client.receive_messages():
        # Process messages
        pass

Message Types

The SDK supports various message types corresponding to different iFlow protocol responses:

AssistantMessage - AI Assistant Response

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def handle_assistant_message():
    async with IFlowClient() as client:
        await client.send_message("Please introduce 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 - Tool Execution

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

async def handle_tool_calls():
    async with IFlowClient() as client:
        # Note: This example demonstrates how to view tool call messages
        # iFlow will request full paths and content to create files
        await client.send_message("List files in current directory")
        
        async for message in client.receive_messages():
            if isinstance(message, ToolCallMessage):
                print(f"Tool: {message.label}")  
            elif isinstance(message, TaskFinishMessage):
                break

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

PlanMessage - Task Planning

import asyncio
from iflow_sdk import IFlowClient, PlanMessage, TaskFinishMessage

async def show_plan():
    async with IFlowClient() as client:
        await client.send_message("Help me create a Python project structure")
        
        async for message in client.receive_messages():
            if isinstance(message, PlanMessage):
                print("Execution Plan:")
                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 - Task Completion

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

async def check_completion():
    async with IFlowClient() as client:
        await client.send_message("Calculate 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()  # New line
                if message.stop_reason == StopReason.END_TURN:
                    print("Task completed successfully")
                elif message.stop_reason == StopReason.MAX_TOKENS:
                    print("Max token limit reached")
                break  # TaskFinishMessage indicates end of conversation

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

Common Use Cases

Interactive Chatbot

#!/usr/bin/env python3
import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def chatbot():
    print("iFlow Chatbot (type 'quit' to exit)")
    print("-" * 50)
    
    async with IFlowClient() as client:
        while True:
            user_input = input("\\nYou: ")
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("Goodbye!")
                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()  # New line
                    break

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

Streaming Response Handling

from iflow_sdk import query_stream
import asyncio

async def stream_example():
    prompt = "Explain what is sublimation phenomenon"
    
    # query_stream returns an async generator of text chunks
    async for chunk in query_stream(prompt):
        print(chunk, end="", flush=True)
    print()  # Final new line

asyncio.run(stream_example())

Advanced Configuration

Manual Process Management

If you need to manually manage iFlow processes:

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

async def manual_process_example():
    # Disable automatic process management
    options = IFlowOptions(
        auto_start_process=False,
        url="ws://localhost:8090/acp"  # Connect to existing iFlow
    )
    
    async with IFlowClient(options) as client:
        await client.send_message("Your question")
        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())

Note Manual mode requires you to start iFlow separately:
iflow --experimental-acp --port 8090

Error Handling

The SDK provides detailed error handling mechanisms:

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("Test")
            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"Connection error: {e}")
    except TimeoutError as e:
        print(f"Timeout error: {e}")
    except Exception as e:
        print(f"Unknown error: {e}")

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

Synchronous Calls

For scenarios requiring synchronous calls:

from iflow_sdk import query_sync, IFlowOptions

# Synchronous call with timeout control
options = IFlowOptions(timeout=30.0)
response = query_sync("Your question", options=options)
print(response)

API Reference

Core Classes

Class	Description
`IFlowClient`	Main client class managing iFlow connections
`IFlowOptions`	Configuration options class
`RawDataClient`	Client for accessing raw protocol data

Message Types

Message Type	Description	Key Properties
`AssistantMessage`	AI assistant text responses	`chunk.text`, `agent_id`
`ToolCallMessage`	Tool execution requests and status	`label`, `status`, `content`, `icon`
`PlanMessage`	Structured task plans	`entries` (containing `content`, `priority`, `status`)
`TaskFinishMessage`	Task completion signals	`stop_reason`

Troubleshooting

Connection Issues

If you encounter connection errors, check:

Is iFlow installed
```
iflow --version
```

Is the port occupied

# Linux/Mac
lsof -i :8090

# Windows
netstat -an | findstr :8090

Manual connection test
```
iflow --experimental-acp --port 8090
```

Timeout Issues

For long-running tasks, increase the timeout:

from iflow_sdk import IFlowClient, IFlowOptions

options = IFlowOptions(timeout=600.0)  # 10 minute timeout
client = IFlowClient(options)

Debug Logging

Enable verbose logging for debugging:

import logging
from iflow_sdk import IFlowClient, IFlowOptions

# Set log level
logging.basicConfig(level=logging.DEBUG)

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

Overview​

System Requirements​

Installation​

Quick Start​

Basic Example​

Simple Query​

Core Concepts​

IFlowClient​

Message Types​

AssistantMessage - AI Assistant Response​

ToolCallMessage - Tool Execution​

PlanMessage - Task Planning​

TaskFinishMessage - Task Completion​

Common Use Cases​

Interactive Chatbot​

Streaming Response Handling​

Advanced Configuration​

Manual Process Management​

Error Handling​

Synchronous Calls​

API Reference​

Core Classes​

Message Types​

Troubleshooting​

Connection Issues​

Timeout Issues​

Debug Logging​

Overview

System Requirements

Installation

Quick Start

Basic Example

Simple Query

Core Concepts

IFlowClient

Message Types

AssistantMessage - AI Assistant Response

ToolCallMessage - Tool Execution

PlanMessage - Task Planning

TaskFinishMessage - Task Completion

Common Use Cases

Interactive Chatbot

Streaming Response Handling

Advanced Configuration

Manual Process Management

Error Handling

Synchronous Calls

API Reference

Core Classes

Message Types

Troubleshooting

Connection Issues

Timeout Issues

Debug Logging