Triggers

Register and manage triggers in the III SDK

Overview

Triggers allow functions to be invoked automatically in response to events. The III SDK supports registering triggers and implementing custom trigger types.

Registering Triggers

register_trigger

Register a trigger that automatically invokes a function when an event occurs.

trigger = iii.register_trigger(
    type="http",
    function_id="api.todos.create",
    config={
        "api_path": "/todos",
        "http_method": "POST"
    }
)

# Later: unregister the trigger
trigger.unregister()

type · str · required

The trigger type ID (e.g., "http", "cron", "engine::functions-available")

function_id · str · required

The function ID to invoke when the trigger fires

config · Any · required

Trigger-specific configuration (varies by trigger type)

trigger · Trigger

Trigger object with unregister() method

Built-in Trigger Types

HTTP Trigger

Exposes a function as an HTTP REST endpoint:

from iii import ApiRequest, ApiResponse

async def handle_request(data):
    req = ApiRequest(**data)
    
    # Access request data
    body = req.body
    path_params = req.path_params
    query_params = req.query_params
    headers = req.headers
    method = req.method
    
    return ApiResponse(
        status_code=200,
        body={"message": "Success"},
        headers={"X-Custom": "value"}
    )

iii.register_function("api.handler", handle_request)

iii.register_trigger(
    type="http",
    function_id="api.handler",
    config={
        "api_path": "/api/endpoint",
        "http_method": "POST",
        "description": "Handle POST requests"
    }
)

Config Fields:

• api_path (str): URL path for the endpoint
• http_method (str): HTTP method ("GET", "POST", "PUT", "PATCH", "DELETE")
• description (str, optional): Human-readable description

Functions Available Trigger

Invoked when functions become available in the engine:

async def on_functions_ready(data):
    functions = data.get("functions", [])
    print(f"Functions available: {len(functions)}")

iii.register_function("init.handler", on_functions_ready)

iii.register_trigger(
    type="engine::functions-available",
    function_id="init.handler",
    config={}
)

Custom Trigger Types

register_trigger_type

Register a custom trigger type with a handler that manages trigger lifecycles.

from iii import TriggerHandler, TriggerConfig

class CronTriggerHandler(TriggerHandler):
    def __init__(self, iii_client):
        self.iii = iii_client
        self.tasks = {}
    
    async def register_trigger(self, config: TriggerConfig) -> None:
        """Called when a trigger of this type is registered."""
        trigger_id = config.id
        function_id = config.function_id
        cron_expr = config.config["expression"]
        
        # Start a background task
        task = asyncio.create_task(
            self._run_cron(trigger_id, function_id, cron_expr)
        )
        self.tasks[trigger_id] = task
    
    async def unregister_trigger(self, config: TriggerConfig) -> None:
        """Called when a trigger of this type is unregistered."""
        trigger_id = config.id
        task = self.tasks.pop(trigger_id, None)
        if task:
            task.cancel()
    
    async def _run_cron(self, trigger_id, function_id, cron_expr):
        # Implement cron scheduling logic
        while True:
            await asyncio.sleep(60)  # Example: every minute
            self.iii.call_void(function_id, {"trigger_id": trigger_id})

# Register the trigger type
handler = CronTriggerHandler(iii)
iii.register_trigger_type(
    id="cron",
    description="Execute functions on a schedule",
    handler=handler
)

# Now users can register cron triggers
iii.register_trigger(
    type="cron",
    function_id="tasks.cleanup",
    config={"expression": "0 0 * * *"}  # Daily at midnight
)

id · str · required

Unique trigger type ID (e.g., "cron", "webhook")

description · str · required

Human-readable description of what this trigger type does

handler · TriggerHandler · required

Handler instance implementing register_trigger() and unregister_trigger()

unregister_trigger_type

Unregister a custom trigger type:

iii.unregister_trigger_type("cron")

id · str · required

The trigger type ID to unregister

Types

Trigger

Represents a registered trigger instance.

class Trigger:
    def unregister(self) -> None:
        """Unregister this trigger."""
        ...

TriggerHandler

Abstract base class for custom trigger type handlers.

from abc import ABC, abstractmethod
from typing import Generic, TypeVar

TConfig = TypeVar("TConfig")

class TriggerHandler(ABC, Generic[TConfig]):
    @abstractmethod
    async def register_trigger(self, config: TriggerConfig[TConfig]) -> None:
        """Register a trigger with the given configuration."""
        pass
    
    @abstractmethod
    async def unregister_trigger(self, config: TriggerConfig[TConfig]) -> None:
        """Unregister a trigger with the given configuration."""
        pass

TriggerConfig

Configuration passed to trigger handlers.

class TriggerConfig(BaseModel, Generic[TConfig]):
    id: str  # Unique trigger instance ID
    function_id: str  # Function to invoke
    config: Any  # Trigger-specific configuration

Example: Webhook Trigger Type

import asyncio
from aiohttp import web
from iii import III, TriggerHandler, TriggerConfig

class WebhookTriggerHandler(TriggerHandler):
    def __init__(self, iii_client, port=8080):
        self.iii = iii_client
        self.port = port
        self.routes = {}  # path -> (trigger_id, function_id)
        self.app = None
        self.runner = None
    
    async def register_trigger(self, config: TriggerConfig) -> None:
        trigger_id = config.id
        function_id = config.function_id
        path = config.config["path"]
        
        self.routes[path] = (trigger_id, function_id)
        
        # Start HTTP server if not already running
        if self.app is None:
            await self._start_server()
    
    async def unregister_trigger(self, config: TriggerConfig) -> None:
        path = config.config["path"]
        self.routes.pop(path, None)
    
    async def _start_server(self):
        self.app = web.Application()
        self.app.router.add_post("/{path:.*}", self._handle_webhook)
        
        self.runner = web.AppRunner(self.app)
        await self.runner.setup()
        site = web.TCPSite(self.runner, "0.0.0.0", self.port)
        await site.start()
        print(f"Webhook server listening on port {self.port}")
    
    async def _handle_webhook(self, request):
        path = request.path
        
        if path not in self.routes:
            return web.Response(status=404, text="Not found")
        
        trigger_id, function_id = self.routes[path]
        
        # Parse request body
        body = await request.json()
        
        # Invoke the function
        try:
            result = await self.iii.call(function_id, {
                "trigger_id": trigger_id,
                "body": body,
                "headers": dict(request.headers)
            })
            return web.json_response(result)
        except Exception as e:
            return web.Response(status=500, text=str(e))

async def main():
    iii = III("ws://localhost:49134")
    await iii.connect()
    
    # Register webhook trigger type
    webhook_handler = WebhookTriggerHandler(iii, port=8080)
    iii.register_trigger_type(
        id="webhook",
        description="Invoke functions via HTTP webhooks",
        handler=webhook_handler
    )
    
    # Register a function to handle webhooks
    async def handle_webhook(data):
        print(f"Webhook received: {data}")
        return {"status": "processed"}
    
    iii.register_function("webhooks.github", handle_webhook)
    
    # Register a webhook trigger
    iii.register_trigger(
        type="webhook",
        function_id="webhooks.github",
        config={"path": "/github"}
    )
    
    # Keep running
    await asyncio.Event().wait()

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

ApiRequest / ApiResponse

Types for HTTP trigger handlers.

ApiRequest

class ApiRequest(BaseModel, Generic[TInput]):
    path_params: dict[str, str]  # URL path parameters
    query_params: dict[str, str | list[str]]  # Query string parameters
    body: Any  # Request body (parsed JSON)
    headers: dict[str, str | list[str]]  # HTTP headers
    method: str  # HTTP method ("GET", "POST", etc.)

ApiResponse

class ApiResponse(BaseModel, Generic[TOutput]):
    status_code: int  # HTTP status code (200, 404, etc.)
    body: Any  # Response body (will be JSON serialized)
    headers: dict[str, str]  # HTTP response headers

Example Usage

from iii import ApiRequest, ApiResponse

async def get_user(data):
    req = ApiRequest(**data)
    
    user_id = req.path_params.get("id")
    include_email = req.query_params.get("include_email") == "true"
    auth_token = req.headers.get("Authorization")
    
    # Fetch user...
    user = {"id": user_id, "name": "Alice"}
    if include_email:
        user["email"] = "[e-mail]"
    
    return ApiResponse(
        status_code=200,
        body=user,
        headers={"Cache-Control": "max-age=300"}
    )

iii.register_function("api.users.get", get_user)

iii.register_trigger(
    type="http",
    function_id="api.users.get",
    config={
        "api_path": "/users/:id",
        "http_method": "GET"
    }
)