Skip to main content

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

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"] = "alice@example.com"
    
    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"
    }
)

Build docs developers (and LLMs) love

Get started for freeTalk to us