> ## 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.
# 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.
```python theme={null}
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:
```python theme={null}
try:
result = await iii.call("users.get", {"id": "nonexistent"})
except Exception as e:
print(f"Error: {e}") # "Error: User not found"
```
### Timeout Errors
```python theme={null}
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:
```python theme={null}
# 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).
```python theme={null}
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
```python theme={null}
# 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:
```python theme={null}
# 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:
```python theme={null}
# 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:
```python theme={null}
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:
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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())
```