Functions

Register and invoke functions in the III SDK

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)

path · str · required

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

description · str

Human-readable description of what the function does

metadata · dict[str, Any]

Additional metadata (e.g., {"version": "1.0", "team": "platform"})

ref · FunctionRef

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

The 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)

id · str · required

Unique function ID

config · HttpInvocationConfig · required

HTTP invocation configuration

ref · FunctionRef

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

url · str · required

The HTTP endpoint URL

method · str · default: "POST"

HTTP method: "GET", "POST", "PUT", "PATCH", or "DELETE"

timeout_ms · int

Request timeout in milliseconds

headers · dict[str, str]

HTTP headers to include in the request

auth · HttpAuthConfig

Authentication configuration (HMAC, Bearer, or API Key)

HTTP Authentication

Bearer Token

from iii import HttpAuthBearer

auth = HttpAuthBearer(token_key="MY_TOKEN_ENV_VAR")

token_key · str · required

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"
)

header · str · required

HTTP header name for the API key

value_key · str · required

Environment variable name containing the API key

HMAC

from iii import HttpAuthHmac

auth = HttpAuthHmac(secret_key="HMAC_SECRET_ENV_VAR")

secret_key · str · required

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"}

path · str · required

The function ID to invoke

data · Any · required

Data to pass to the function

timeout · float

Timeout in seconds

result · Any

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"})

path · str · required

The function ID to invoke

data · Any · required

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": "[e-mail]"})
    print(f"Created user: {user}")
    
    await asyncio.Event().wait()

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