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.
Function Registration
register_function
Register a function handler that can be invoked by other workers.
async def greet(data):
name = data.get("name", "World")
return {"message": f"Hello, {name}!"}
ref = iii.register_function("greet", greet)
Unique function ID (e.g., "users.create", "orders.process")
handler
Callable[[Any], Awaitable[Any]]
required
Async function that receives invocation data and returns a result
Human-readable description of what the function does
Additional metadata (e.g., {"version": "1.0", "team": "platform"})
Reference object with id and unregister() method
Handler Signature
Function handlers must be async functions:
from typing import Any
async def my_handler(data: Any) -> Any:
"""Process the input data and return a result."""
# Your logic here
return result
data parameter contains the invocation payload. Return values are automatically serialized.
Unregistering Functions
# Option 1: Using the returned reference
ref = iii.register_function("my.function", handler)
ref.unregister()
# Option 2: Using the reference ID
ref.id # "my.function"
HTTP Functions
register_http_function
Register a function that invokes an external HTTP endpoint instead of a Python handler.
from iii import HttpInvocationConfig, HttpAuthBearer
config = HttpInvocationConfig(
url="https://api.example.com/process",
method="POST",
timeout_ms=30000,
headers={"Content-Type": "application/json"},
auth=HttpAuthBearer(token_key="BEARER_TOKEN_ENV_VAR")
)
ref = iii.register_http_function("external.process", config)
config
HttpInvocationConfig
required
HTTP invocation configuration
Reference object with id and unregister() method
HttpInvocationConfig
class HttpInvocationConfig(BaseModel):
url: str
method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST"
timeout_ms: int | None = None
headers: dict[str, str] | None = None
auth: HttpAuthConfig | None = None
HTTP method: "GET", "POST", "PUT", "PATCH", or "DELETE"
Request timeout in milliseconds
HTTP headers to include in the request
Authentication configuration (HMAC, Bearer, or API Key)
HTTP Authentication
Bearer Token
from iii import HttpAuthBearer
auth = HttpAuthBearer(token_key="MY_TOKEN_ENV_VAR")
Environment variable name containing the bearer token
API Key
from iii import HttpAuthApiKey
auth = HttpAuthApiKey(
header="X-API-Key",
value_key="API_KEY_ENV_VAR"
)
HTTP header name for the API key
Environment variable name containing the API key
HMAC
from iii import HttpAuthHmac
auth = HttpAuthHmac(secret_key="HMAC_SECRET_ENV_VAR")
Environment variable name containing the HMAC secret
Function Invocation
call
Invoke a function and await the response.
result = await iii.call("orders.process", {"order_id": "123"})
print(result) # {"status": "processed"}
The function ID to invoke
Data to pass to the function
The function’s return value
call_void
Invoke a function without waiting for a response (fire-and-forget).
iii.call_void("notifications.send", {"user_id": "456", "message": "Hello"})
The function ID to invoke
Data to pass to the function
trigger / trigger_void
Aliases for call and call_void:
# Same as call()
result = await iii.trigger("my.function", data)
# Same as call_void()
iii.trigger_void("my.function", data)
Error Handling
Functions can raise exceptions, which are propagated to the caller:
async def divide(data):
a = data["a"]
b = data["b"]
if b == 0:
raise ValueError("Division by zero")
return {"result": a / b}
iii.register_function("math.divide", divide)
# Calling with invalid data
try:
result = await iii.call("math.divide", {"a": 10, "b": 0})
except Exception as e:
print(f"Error: {e}") # "Error: Division by zero"
Example: Service Architecture
import asyncio
from iii import III, get_context
iii = III("ws://localhost:49134")
# User service
async def create_user(data):
ctx = get_context()
ctx.logger.info("Creating user", data={"email": data["email"]})
# Save to database...
user_id = "user123"
# Trigger welcome email
iii.call_void("email.send_welcome", {"user_id": user_id})
return {"id": user_id, "email": data["email"]}
# Email service
async def send_welcome_email(data):
ctx = get_context()
user_id = data["user_id"]
# Fetch user details
user = await iii.call("users.get", {"id": user_id})
ctx.logger.info(f"Sending welcome email to {user['email']}")
# Send email...
return {"status": "sent"}
iii.register_function(
"users.create",
create_user,
description="Create a new user account",
metadata={"service": "users", "version": "1.0"}
)
iii.register_function(
"email.send_welcome",
send_welcome_email,
description="Send welcome email to new users",
metadata={"service": "email"}
)
async def main():
await iii.connect()
# Test the flow
user = await iii.call("users.create", {"email": "alice@example.com"})
print(f"Created user: {user}")
await asyncio.Event().wait()
if __name__ == "__main__":
asyncio.run(main())