Invocation

Call and trigger functions in the III SDK

Overview

The III SDK provides methods to invoke functions synchronously (awaiting a response) or asynchronously (fire-and-forget).

Synchronous Invocation

call

Invoke a function and await the response.

result = await iii.call("users.get", {"id": "123"})
print(result)  # {"id": "123", "name": "Alice"}

path · str · required

The function ID to invoke

data · Any · required

Data to pass to the function (typically a dict)

timeout · float

Timeout in seconds. Raises TimeoutError if exceeded

result · Any

The function’s return value

Error Handling

Exceptions from the remote function are propagated to the caller:

try:
    result = await iii.call("users.get", {"id": "nonexistent"})
except Exception as e:
    print(f"Error: {e}")  # "Error: User not found"

Timeout Errors

import asyncio

try:
    result = await iii.call("slow.function", {}, timeout=5.0)
except TimeoutError:
    print("Function timed out after 5 seconds")

Custom Timeout

Override the default timeout on a per-call basis:

# Short timeout for health checks
status = await iii.call("health.check", {}, timeout=1.0)

# Long timeout for batch processing
result = await iii.call("batch.process", {"items": items}, timeout=300.0)

Asynchronous Invocation

call_void

Invoke a function without waiting for a response (fire-and-forget).

iii.call_void("notifications.send", {
    "user_id": "456",
    "message": "Your order has shipped"
})

print("Notification queued")  # Returns immediately

path · str · required

The function ID to invoke

data · Any · required

Data to pass to the function

Use Cases

call_void is ideal for:

• Notifications: Sending emails, SMS, push notifications
• Logging: Fire-and-forget audit logs
• Background tasks: Queue jobs that don’t need immediate results
• Event broadcasting: Notify multiple subscribers

# Audit logging
iii.call_void("audit.log", {
    "user_id": user_id,
    "action": "user.login",
    "timestamp": time.time()
})

# Event broadcasting
iii.call_void("events.publish", {
    "topic": "order.created",
    "data": order_data
})

Aliases

trigger

Alias for call(). Both methods are equivalent:

# These are identical
result = await iii.call("my.function", data)
result = await iii.trigger("my.function", data)

trigger_void

Alias for call_void(). Both methods are equivalent:

# These are identical
iii.call_void("my.function", data)
iii.trigger_void("my.function", data)

Distributed Tracing

The SDK automatically propagates OpenTelemetry trace context across function calls when OTel is initialized:

from iii import init_otel
from opentelemetry import trace

init_otel()
tracer = trace.get_tracer(__name__)

async def parent_function(data):
    with tracer.start_as_current_span("parent-operation"):
        # Trace context is automatically propagated
        result = await iii.call("child.function", data)
        return result

async def child_function(data):
    # This span is linked to the parent trace
    with tracer.start_as_current_span("child-operation"):
        # Do work
        return {"result": "success"}

iii.register_function("parent.function", parent_function)
iii.register_function("child.function", child_function)

Trace context is propagated via W3C Trace Context headers (traceparent and baggage).

Channel References

The SDK automatically resolves StreamChannelRef objects into ChannelReader or ChannelWriter instances:

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

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

iii.register_function("producer", producer)
iii.register_function("consumer", consumer)

Example: Request-Response Pattern

import asyncio
from iii import III

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

# Service A: Order processing
async def process_order(data):
    order_id = data["order_id"]
    
    # Validate inventory
    inventory = await iii.call("inventory.check", {
        "product_id": data["product_id"],
        "quantity": data["quantity"]
    })
    
    if not inventory["available"]:
        raise ValueError("Product out of stock")
    
    # Process payment
    payment = await iii.call("payment.charge", {
        "amount": data["amount"],
        "customer_id": data["customer_id"]
    }, timeout=60.0)  # Long timeout for payment processing
    
    # Send confirmation (fire-and-forget)
    iii.call_void("email.send", {
        "to": data["email"],
        "template": "order_confirmation",
        "data": {"order_id": order_id}
    })
    
    return {
        "order_id": order_id,
        "status": "confirmed",
        "payment_id": payment["id"]
    }

# Service B: Inventory management
async def check_inventory(data):
    product_id = data["product_id"]
    quantity = data["quantity"]
    
    # Check database...
    available_qty = 100  # Example
    
    return {
        "available": available_qty >= quantity,
        "quantity": available_qty
    }

# Service C: Payment processing
async def charge_payment(data):
    # Process payment...
    return {
        "id": "payment_123",
        "status": "success",
        "amount": data["amount"]
    }

# Service D: Email notifications
async def send_email(data):
    print(f"Sending email to {data['to']} with template {data['template']}")
    # Send email...
    return {"status": "sent"}

iii.register_function("orders.process", process_order)
iii.register_function("inventory.check", check_inventory)
iii.register_function("payment.charge", charge_payment)
iii.register_function("email.send", send_email)

async def main():
    await iii.connect()
    
    # Test the flow
    try:
        result = await iii.call("orders.process", {
            "order_id": "order_789",
            "product_id": "prod_123",
            "quantity": 2,
            "amount": 49.99,
            "customer_id": "cust_456",
            "email": "[e-mail]"
        })
        print(f"Order processed: {result}")
    except Exception as e:
        print(f"Order failed: {e}")
    
    await asyncio.Event().wait()

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

Example: Event-Driven Architecture

import asyncio
from iii import III

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

# Event publisher
async def publish_event(data):
    event_type = data["type"]
    event_data = data["data"]
    
    # Get all subscribers for this event type
    subscribers = await iii.call("events.subscribers", {"type": event_type})
    
    # Notify all subscribers (fire-and-forget)
    for subscriber in subscribers["functions"]:
        iii.call_void(subscriber, {
            "event_type": event_type,
            "data": event_data
        })
    
    return {"notified": len(subscribers["functions"])}

# Subscriber 1: Analytics
async def track_analytics(data):
    print(f"Analytics: {data['event_type']} - {data['data']}")
    # Send to analytics service...
    return {"status": "tracked"}

# Subscriber 2: Notifications
async def send_notification(data):
    print(f"Notification: {data['event_type']} - {data['data']}")
    # Send notification...
    return {"status": "sent"}

iii.register_function("events.publish", publish_event)
iii.register_function("analytics.track", track_analytics)
iii.register_function("notifications.send", send_notification)

async def main():
    await iii.connect()
    
    # Publish an event
    result = await iii.call("events.publish", {
        "type": "user.signup",
        "data": {"user_id": "123", "email": "[e-mail]"}
    })
    print(f"Event published to {result['notified']} subscribers")
    
    await asyncio.Event().wait()

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