Quickstart

pip install iii-sdk

import asyncio
from iii import III

async def greet(data):
    name = data.get("name", "World")
    return {"message": f"Hello, {name}!"}

iii = III("ws://localhost:49134")
iii.register_function("greet", greet)

async def main():
    await iii.connect()
    
    # Call the function
    result = await iii.call("greet", {"name": "Alice"})
    print(result)  # {"message": "Hello, Alice!"}
    
    # Keep running
    await asyncio.Event().wait()

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

from iii import III

iii = III("ws://localhost:49134")

async def process_order(data):
    order_id = data["order_id"]
    # Process the order...
    return {"status": "processed", "order_id": order_id}

iii.register_function(
    "orders.process",
    process_order,
    description="Process a customer order",
    metadata={"version": "1.0", "team": "orders"}
)

# Await response (default 30s timeout)
result = await iii.call("orders.process", {"order_id": "123"})

# Custom timeout
result = await iii.call("orders.process", {"order_id": "123"}, timeout=60.0)

# Fire-and-forget (no response)
iii.call_void("notifications.send", {"user_id": "456", "message": "Order shipped"})

from iii import get_context

async def process_payment(data):
    ctx = get_context()
    ctx.logger.info("Processing payment", data={"amount": data["amount"]})
    
    # Process payment...
    
    ctx.logger.info("Payment processed successfully")
    return {"status": "success"}

from iii import III, ApiRequest, ApiResponse

iii = III("ws://localhost:49134")

async def create_todo(data):
    req = ApiRequest(**data)
    title = req.body.get("title")
    
    # Save todo...
    
    return ApiResponse(
        status_code=201,
        body={"id": "123", "title": title}
    )

iii.register_function("api.todos.create", create_todo)

async def main():
    await iii.connect()
    
    # Register HTTP trigger
    iii.register_trigger(
        type="http",
        function_id="api.todos.create",
        config={
            "api_path": "/todos",
            "http_method": "POST"
        }
    )
    
    await asyncio.Event().wait()

asyncio.run(main())

async def producer(data):
    channel = await iii.create_channel()
    
    # Pass writer_ref to another function
    iii.call_void("consumer", {"reader": channel.reader_ref})
    
    # Write data
    await channel.writer.write(b"chunk1")
    await channel.writer.write(b"chunk2")
    await channel.writer.close_async()
    
    return {"status": "sent"}

async def consumer(data):
    reader = data["reader"]  # Automatically resolved to ChannelReader
    
    async for chunk in reader:
        print(f"Received: {chunk}")
    
    return {"status": "received"}

from iii import III, InitOptions, ReconnectionConfig

options = InitOptions(
    worker_name="my-worker",
    invocation_timeout_ms=60000,  # 60 seconds
    reconnection_config=ReconnectionConfig(
        initial_delay_ms=2000,
        max_delay_ms=60000,
        backoff_multiplier=2.0,
        max_retries=-1  # Infinite retries
    )
)

iii = III("ws://localhost:49134", options)

async def main():
    iii = III("ws://localhost:49134")
    await iii.connect()
    
    try:
        # Your application logic
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        await iii.shutdown()

asyncio.run(main())