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
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"}
The function ID to invoke
Data to pass to the function (typically a dict)
Timeout in seconds. Raises TimeoutError if exceeded
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
The function ID to invoke
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)
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": "customer@example.com"
})
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": "user@example.com"}
})
print(f"Event published to {result['notified']} subscribers")
await asyncio.Event().wait()
if __name__ == "__main__":
asyncio.run(main())