We are going to build an event-driven router that ingests JSON events and uses an LLM to select and parameterize the correct handler function. This pattern replaces brittle if-else chains with a model that can reason over unstructured payloads, which is useful for anyone building automation around webhooks, logs, or message queues. I will use Oxlo.ai because its flat per-request pricing keeps costs predictable even when event payloads are large or the system prompt grows.
What you'll need
- Python 3.10 or newer
pip install openai- An Oxlo.ai API key from https://portal.oxlo.ai
Step 1: Define the handlers
Create a registry of plain Python functions that represent your business logic. The LLM will later choose which one to call and what arguments to pass.
# agent.py
from typing import Callable
def send_welcome_email(user_id: str, email: str) -> None:
print(f"[HANDLER] Sending welcome email to {email} (user {user_id})")
def alert_ops_team(service: str, severity: str, message: str) -> None:
print(f"[HANDLER] Alerting ops: {service} | {severity} | {message}")
def retry_payment(transaction_id: str, amount: float) -> None:
print(f"[HANDLER] Retrying payment {transaction_id} for ${amount}")
HANDLER_REGISTRY: dict[str, Callable[..., None]] = {
"send_welcome_email": send_welcome_email,
"alert_ops_team": alert_ops_team,
"retry_payment": retry_payment,
}Step 2: Build the event stream
Add a mock generator that yields realistic events. In production you would swap this for a Kafka topic, webhook endpoint, or message broker, but a generator is enough to test the loop locally.
# agent.py
from typing import Callable, Iterator
import json
# --- Handlers ---
def send_welcome_email(user_id: str, email: str) -> None:
print(f"[HANDLER] Sending welcome email to {email} (user {user_id})")
def alert_ops_team(service: str, severity: str, message: str) -> None:
print(f"[HANDLER] Alerting ops: {service} | {severity} | {message}")
def retry_payment(transaction_id: str, amount: float) -> None:
print(f"[HANDLER] Retrying payment {transaction_id} for ${amount}")
HANDLER_REGISTRY: dict[str, Callable[..., None]] = {
"send_welcome_email": send_welcome_email,
"alert_ops_team": alert_ops_team,
"retry_payment": retry_payment,
}
# --- Event stream ---
EVENTS = [
{"event_type": "user_signup", "payload": {"user_id": "u-8821", "email": "dev@example.com"}},
{"event_type": "payment_failed", "payload": {"transaction_id": "tx-9912", "amount": 49.99, "currency": "USD"}},
{"event_type": "server_error", "payload": {"service": "api-gateway", "severity": "critical", "message": "Connection timeout to database"}},
]
def event_stream() -> Iterator[dict]:
for evt in EVENTS:
yield evtStep 3: Write the system prompt
The system prompt is the contract between your code and the model. It lists the available handlers, their signatures, and the exact JSON schema the router expects back. Keep it editable; as you add handlers, update this string.
# agent.py
from typing import Callable, Iterator
import json
# --- System prompt ---
SYSTEM_PROMPT = """You are the routing layer of an event-driven system.
You receive a JSON event. Your job is to select exactly one handler from the catalog below and provide the arguments it needs.
Handler catalog:
- send_welcome_email(user_id: str, email: str)
- alert_ops_team(service: str, severity: str, message: str)
- retry_payment(transaction_id: str, amount: float)
Rules:
1. Respond ONLY with a JSON object. No markdown, no explanation.
2. The JSON must have two keys: "handler" (string) and "args" (object mapping parameter names to values).
3. Extract values from the event payload. If a required argument is missing, use an empty string or 0.
4. Do not invent handlers that are not in the catalog.
Example input: {"event_type": "user_signup", "payload": {"user_id": "123", "email": "a@b.com"}}
Example output: {"handler": "send_welcome_email", "args": {"user_id": "123", "email": "a@b.com"}}"""
# --- Handlers ---
def send_welcome_email(user_id: str, email: str) -> None:
print(f"[HANDLER] Sending welcome email to {email} (user {user_id})")
def alert_ops_team(service: str, severity: str, message: str) -> None:
print(f"[HANDLER] Alerting ops: {service} | {severity} | {message}")
def retry_payment(transaction_id: str, amount: float) -> None:
print(f"[HANDLER] Retrying payment {transaction_id} for ${amount}")
HANDLER_REGISTRY: dict[str, Callable[..., None]] = {
"send_welcome_email": send_welcome_email,
"alert_ops_team": alert_ops_team,
"retry_payment": retry_payment,
}
# --- Event stream ---
EVENTS = [
{"event_type": "user_signup", "payload": {"user_id": "u-8821", "email": "dev@example.com"}},
{"event_type": "payment_failed", "payload": {"transaction_id": "tx-9912", "amount": 49.99, "currency": "USD"}},
{"event_type": "server_error", "payload": {"service": "api-gateway", "severity": "critical", "message": "Connection timeout to database"}},
]
def event_stream() -> Iterator[dict]:
for evt in EVENTS:
yield evtStep 4: Create the router
Initialize the OpenAI SDK pointing at Oxlo.ai and implement the route_event function. It serializes the event, calls llama-3.3-70b, and safely executes the chosen handler. Oxlo.ai is fully OpenAI-compatible, so the client code is a drop-in replacement.
# agent.py
from typing import Callable, Iterator
import json
from openai import OpenAI
# --- System prompt ---
SYSTEM_PROMPT = """You are the routing layer of an event-driven system.
You receive a JSON event. Your job is to select exactly one handler from the catalog below and provide the arguments it needs.
Handler catalog:
- send_welcome_email(user_id: str, email: str)
- alert_ops_team(service: str, severity: str, message: str)
- retry_payment(transaction_id: str, amount: float)
Rules:
1. Respond ONLY with a JSON object. No markdown, no explanation.
2. The JSON must have two keys: "handler" (string) and "args" (object mapping parameter names to values).
3. Extract values from the event payload. If a required argument is missing, use an empty string or 0.
4. Do not invent handlers that are not in the catalog.
Example input: {"event_type": "user_signup", "payload": {"user_id": "123", "email": "a@b.com"}}
Example output: {"handler": "send_welcome_email", "args": {"user_id": "123", "email": "a@b.com"}}"""
# --- Handlers ---
def send_welcome_email(user_id: str, email: str) -> None:
print(f"[HANDLER] Sending welcome email to {email} (user {user_id})")
def alert_ops_team(service: str, severity: str, message: str) -> None:
print(f"[HANDLER] Alerting ops: {service} | {severity} | {message}")
def retry_payment(transaction_id: str, amount: float) -> None:
print(f"[HANDLER] Retrying payment {transaction_id} for ${amount}")
HANDLER_REGISTRY: dict[str, Callable[..., None]] = {
"send_welcome_email": send_welcome_email,
"alert_ops_team": alert_ops_team,
"retry_payment": retry_payment,
}
# --- Event stream ---
EVENTS = [
{"event_type": "user_signup", "payload": {"user_id": "u-8821", "email": "dev@example.com"}},
{"event_type": "payment_failed", "payload": {"transaction_id": "tx-9912", "amount": 49.99, "currency": "USD"}},
{"event_type": "server_error", "payload": {"service": "api-gateway", "severity": "critical", "message": "Connection timeout to database"}},
]
def event_stream() -> Iterator[dict]:
for evt in EVENTS:
yield evt
# --- Router ---
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def route_event(event: dict) -> None:
event_json = json.dumps(event, ensure_ascii=False)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": event_json},
],
)
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("```
")[1].replace("json", "").strip()
decision = json.loads(raw)
handler_name = decision["handler"]
args = decision["args"]
if handler_name not in HANDLER_REGISTRY:
raise ValueError(f"Unknown handler: {handler_name}")
HANDLER_REGISTRY[handler_name](**args)Step 5: Wire the event loop
Add the entrypoint that consumes the stream, prints the event type, and lets the router dispatch it. A small try-except block keeps one bad event from killing the process.
# agent.py
from typing import Callable, Iterator
import json
from openai import OpenAI
# --- System prompt ---
SYSTEM_PROMPT = """You are the routing layer of an event-driven system.
You receive a JSON event. Your job is to select exactly one handler from the catalog below and provide the arguments it needs.
Handler catalog:
- send_welcome_email(user_id: str, email: str)
- alert_ops_team(service: str, severity: str, message: str)
- retry_payment(transaction_id: str, amount: float)
Rules:
1. Respond ONLY with a JSON object. No markdown, no explanation.
2. The JSON must have two keys: "handler" (string) and "args" (object mapping parameter names to values).
3. Extract values from the event payload. If a required argument is missing, use an empty string or 0.
4. Do not invent handlers that are not in the catalog.
Example input: {"event_type": "user_signup", "payload": {"user_id": "123", "email": "a@b.com"}}
Example output: {"handler": "send_welcome_email", "args": {"user_id": "123", "email": "a@b.com"}}"""
# --- Handlers ---
def send_welcome_email(user_id: str, email: str) -> None:
print(f"[HANDLER] Sending welcome email to {email} (user {user_id})")
def alert_ops_team(service: str, severity: str, message: str) -> None:
print(f"[HANDLER] Alerting ops: {service} | {severity} | {message}")
def retry_payment(transaction_id: str, amount: float) -> None:
print(f"[HANDLER] Retrying payment {transaction_id} for ${amount}")
HANDLER_REGISTRY: dict[str, Callable[..., None]] = {
"send_welcome_email": send_welcome_email,
"alert_ops_team": alert_ops_team,
"retry_payment": retry_payment,
}
# --- Event stream ---
EVENTS = [
{"event_type": "user_signup", "payload": {"user_id": "u-8821", "email": "dev@example.com"}},
{"event_type": "payment_failed", "payload": {"transaction_id": "tx-9912", "amount": 49.99, "currency": "USD"}},
{"event_type": "server_error", "payload": {"service": "api-gateway", "severity": "critical", "message": "Connection timeout to database"}},
]
def event_stream() -> Iterator[dict]:
for evt in EVENTS:
yield evt
# --- Router ---
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def route_event(event: dict) -> None:
event_json = json.dumps(event, ensure_ascii=False)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": event_json},
],
)
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("```
")[1].replace("json", "").strip()
decision = json.loads(raw)
handler_name = decision["handler"]
args = decision["args"]
if handler_name not in HANDLER_REGISTRY:
raise ValueError(f"Unknown handler: {handler_name}")
HANDLER_REGISTRY[handler_name](**args)
# --- Event loop ---
if __name__ == "__main__":
for event in event_stream():
print(f"[EVENT] {event['event_type']}")
try:
route_event(event)
except Exception as e:
print(f"[ERROR] {e}")
print()Run it
Save the file, replace YOUR_OXLO_API_KEY with your key from the Oxlo.ai portal, and run the script.
python agent.pyYou should see output similar to this:
[EVENT] user_signup
[HANDLER] Sending welcome email to dev@example.com (user u-8821)
[EVENT] payment_failed
[HANDLER] Retrying payment tx-9912 for $49.99
[EVENT] server_error
[HANDLER] Alerting ops: api-gateway | critical | Connection timeout to database
Next steps
That is the core of an LLM-powered event processor. Because Oxlo.ai bills per request rather than per token, you can pass full event contexts and detailed prompts without watching metered costs scale with payload size. See https://oxlo.ai/pricing for plan details. If you want to expand this, swap the mock generator for Redis Streams or RabbitMQ, or add Pydantic validation on the model output before execution.
Top comments (0)