> ## 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.
# 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.
```python theme={null}
trigger = iii.register_trigger(
type="http",
function_id="api.todos.create",
config={
"api_path": "/todos",
"http_method": "POST"
}
)
# Later: unregister the trigger
trigger.unregister()
```
The trigger type ID (e.g., `"http"`, `"cron"`, `"engine::functions-available"`)
The function ID to invoke when the trigger fires
Trigger-specific configuration (varies by trigger type)
Trigger object with `unregister()` method
## Built-in Trigger Types
### HTTP Trigger
Exposes a function as an HTTP REST endpoint:
```python theme={null}
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:
```python theme={null}
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.
```python theme={null}
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
)
```
Unique trigger type ID (e.g., `"cron"`, `"webhook"`)
Human-readable description of what this trigger type does
Handler instance implementing `register_trigger()` and `unregister_trigger()`
### unregister\_trigger\_type
Unregister a custom trigger type:
```python theme={null}
iii.unregister_trigger_type("cron")
```
The trigger type ID to unregister
## Types
### Trigger
Represents a registered trigger instance.
```python theme={null}
class Trigger:
def unregister(self) -> None:
"""Unregister this trigger."""
...
```
### TriggerHandler
Abstract base class for custom trigger type handlers.
```python theme={null}
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.
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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"
}
)
```