Documentation Index
Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
Use this file to discover all available pages before exploring further.
Overview
Channels provide WebSocket-backed streams for worker-to-worker data transfer. They enable streaming large payloads, real-time data processing, and bidirectional communication.
Creating Channels
create_channel
Create a streaming channel pair with a writer and reader.
channel = await iii.create_channel(buffer_size=64)
print(channel.writer) # ChannelWriter instance
print(channel.reader) # ChannelReader instance
print(channel.writer_ref) # Serializable StreamChannelRef
print(channel.reader_ref) # Serializable StreamChannelRef
Optional buffer size for the channel
Channel object containing writer, reader, and their serializable references
Channel Type
@dataclass
class Channel:
writer: ChannelWriter # Write binary data
reader: ChannelReader # Read binary data
writer_ref: StreamChannelRef # Pass to other functions
reader_ref: StreamChannelRef # Pass to other functions
ChannelWriter
WebSocket-backed writer for streaming binary data and text messages.
write
Write binary data to the channel.
writer = channel.writer
await writer.write(b"Hello, world!")
await writer.write(b"More data...")
Binary data to write. Large payloads are automatically chunked (64KB per frame)
send_message_async
Send a text message to the channel.
await writer.send_message_async('{"type": "metadata", "value": 123}')
send_message
Fire-and-forget text message (non-async).
writer.send_message('{"type": "status", "value": "processing"}')
close_async
Close the writer connection.
await writer.close_async()
close
Fire-and-forget close (non-async).
stream Property
Access the writer as a WritableStream interface:
stream = writer.stream
stream.write(b"data") # Fire-and-forget write
stream.end(b"final data") # Write and close
ChannelReader
WebSocket-backed reader for streaming binary data and text messages.
Async Iteration
Read binary chunks using async iteration:
reader = channel.reader
async for chunk in reader:
print(f"Received {len(chunk)} bytes")
# Process chunk...
read_all
Read the entire stream into a single bytes object:
data = await reader.read_all()
print(f"Total size: {len(data)} bytes")
All data from the stream concatenated together
on_message
Register a callback for text messages:
def handle_message(msg: str):
print(f"Message: {msg}")
reader.on_message(handle_message)
# Now iterate over binary data
async for chunk in reader:
# handle_message is called for text messages
# this loop only yields binary data
process(chunk)
callback
Callable[[str], Any]
required
Function called for each text message received
close_async
Close the reader connection.
await reader.close_async()
stream Property
Access the reader as a ReadableStream interface:
stream = reader.stream
async for chunk in stream:
print(chunk)
Passing Channels Between Functions
Channels can be passed between functions using their serializable references:
async def producer(data):
# Create a channel
channel = await iii.create_channel()
# Pass the reader reference to a consumer
iii.call_void("consumer", {"reader": channel.reader_ref})
# Write data
for i in range(10):
await channel.writer.write(f"Chunk {i}\n".encode())
await asyncio.sleep(0.1)
await channel.writer.close_async()
return {"status": "sent"}
async def consumer(data):
# The reader_ref is automatically resolved to a ChannelReader
reader = data["reader"]
chunks = []
async for chunk in reader:
chunks.append(chunk.decode())
print(f"Received: {''.join(chunks)}")
return {"status": "received", "chunks": len(chunks)}
iii.register_function("producer", producer)
iii.register_function("consumer", consumer)
# Start the producer
await iii.call("producer", {})
Streaming Patterns
Producer-Consumer
async def stream_logs(data):
channel = await iii.create_channel()
# Start consumer in background
iii.call_void("process_logs", {"reader": channel.reader_ref})
# Stream log lines
with open("/var/log/app.log", "rb") as f:
while chunk := f.read(8192):
await channel.writer.write(chunk)
await channel.writer.close_async()
return {"status": "streaming_complete"}
async def process_logs(data):
reader = data["reader"]
line_buffer = b""
async for chunk in reader:
line_buffer += chunk
while b"\n" in line_buffer:
line, line_buffer = line_buffer.split(b"\n", 1)
# Process line
print(f"Log: {line.decode()}")
return {"status": "processed"}
Bidirectional Communication
async def chat_handler(data):
# Create two channels for bidirectional communication
to_client = await iii.create_channel()
from_client = await iii.create_channel()
# Pass both channels to client
iii.call_void("chat_client", {
"input": to_client.reader_ref,
"output": from_client.writer_ref
})
# Send messages to client
await to_client.writer.write(b"Welcome!\n")
# Read responses from client
async for msg in from_client.reader:
print(f"Client said: {msg.decode()}")
# Echo back
await to_client.writer.write(b"Echo: " + msg)
await to_client.writer.close_async()
return {"status": "chat_ended"}
async def chat_client(data):
input_reader = data["input"]
output_writer = data["output"]
# Read from input, write to output
async for msg in input_reader:
print(f"Server said: {msg.decode()}")
response = input("You: ").encode() + b"\n"
await output_writer.write(response)
await output_writer.close_async()
return {"status": "done"}
File Upload
async def upload_file(data):
channel = await iii.create_channel(buffer_size=128)
# Pass writer to uploader
upload_task = iii.call("file_uploader", {
"filename": "data.bin",
"reader": channel.reader_ref
})
# Stream file contents
with open("local_file.bin", "rb") as f:
while chunk := f.read(64 * 1024): # 64KB chunks
await channel.writer.write(chunk)
await channel.writer.close_async()
# Wait for upload confirmation
result = await upload_task
return result
async def file_uploader(data):
filename = data["filename"]
reader = data["reader"]
total_bytes = 0
with open(f"/uploads/{filename}", "wb") as f:
async for chunk in reader:
f.write(chunk)
total_bytes += len(chunk)
return {
"status": "uploaded",
"filename": filename,
"size": total_bytes
}
async def transform_data(data):
input_channel = await iii.create_channel()
output_channel = await iii.create_channel()
# Chain transformations
iii.call_void("uppercase_transform", {
"input": input_channel.reader_ref,
"output": output_channel.writer_ref
})
# Send data
await input_channel.writer.write(b"hello world\n")
await input_channel.writer.write(b"streaming data\n")
await input_channel.writer.close_async()
# Read transformed results
result = await output_channel.reader.read_all()
return {"transformed": result.decode()}
async def uppercase_transform(data):
reader = data["input"]
writer = data["output"]
async for chunk in reader:
transformed = chunk.upper()
await writer.write(transformed)
await writer.close_async()
return {"status": "complete"}
HTTP Response Streaming
Use channels with HTTP triggers for streaming responses:
from iii import HttpResponse
async def stream_logs_http(data):
# data contains an embedded response channel
reader = data["request_body"]
writer = data["response"]
response = HttpResponse(writer)
await response.status(200)
await response.headers({"Content-Type": "text/plain"})
# Stream log lines
with open("/var/log/app.log", "rb") as f:
while line := f.readline():
await response.stream.write(line)
response.close()
return {"status": "streamed"}
WritableStream
Node.js-style writable stream interface:
class WritableStream:
def write(self, data: bytes) -> None:
"""Fire-and-forget binary write."""
def end(self, data: bytes | None = None) -> None:
"""Write optional final data and close the stream."""
Usage
writer = channel.writer
stream = writer.stream
stream.write(b"chunk 1")
stream.write(b"chunk 2")
stream.end(b"final chunk")
ReadableStream
Node.js-style readable stream interface:
class ReadableStream:
async def __aiter__(self) -> AsyncIterator[bytes]:
"""Async iteration over binary chunks."""
Usage
reader = channel.reader
stream = reader.stream
async for chunk in stream:
print(chunk)
StreamChannelRef
Serializable reference to a channel endpoint:
class StreamChannelRef(BaseModel):
channel_id: str
access_key: str
direction: Literal["read", "write"]
ChannelReader or ChannelWriter when passed in function invocations.
Error Handling
try:
channel = await iii.create_channel()
await channel.writer.write(b"data")
await channel.writer.close_async()
except Exception as e:
print(f"Channel error: {e}")
Best Practices
- Always close channels: Call
close_async() or close() when done writing
- Handle errors: Wrap channel operations in try-except blocks
- Use appropriate buffer sizes: Larger buffers for high-throughput scenarios
- Chunking: Large payloads are automatically chunked at 64KB per frame
- Text vs Binary: Use
write() for binary data, send_message() for text
Example: Image Processing Pipeline
import asyncio
from iii import III
iii = III("ws://localhost:49134")
async def download_image(data):
url = data["url"]
channel = await iii.create_channel(buffer_size=256)
# Start processing pipeline
process_task = iii.call("process_image", {
"reader": channel.reader_ref,
"format": "thumbnail"
})
# Simulate downloading image
image_data = b"...binary image data..."
await channel.writer.write(image_data)
await channel.writer.close_async()
# Wait for processed result
result = await process_task
return result
async def process_image(data):
reader = data["reader"]
format_type = data["format"]
# Read entire image
image_data = await reader.read_all()
# Process image (resize, compress, etc.)
processed = process_image_data(image_data, format_type)
# Create output channel
output_channel = await iii.create_channel()
# Upload processed image
iii.call_void("upload_image", {
"reader": output_channel.reader_ref,
"filename": f"thumbnail.{format_type}"
})
await output_channel.writer.write(processed)
await output_channel.writer.close_async()
return {"status": "processed", "format": format_type}
async def upload_image(data):
reader = data["reader"]
filename = data["filename"]
image_data = await reader.read_all()
# Upload to storage
url = save_to_storage(filename, image_data)
return {"url": url, "size": len(image_data)}
iii.register_function("download_image", download_image)
iii.register_function("process_image", process_image)
iii.register_function("upload_image", upload_image)
async def main():
await iii.connect()
result = await iii.call("download_image", {
"url": "https://example.com/image.jpg"
})
print(f"Image processed: {result}")
if __name__ == "__main__":
asyncio.run(main())