Event-Driven Agents

Run a Chidori agent as a persistent HTTP server with chidori serve. Every incoming request is converted into an event dict and passed to your agent(event) function. The return value becomes the HTTP response.

chidori serve agents/webhook.star --port 8080

This is how you build webhook handlers, chatbots, alert responders, and any agent that reacts to the outside world rather than being invoked ad-hoc.

The event dict

Every HTTP request is converted to a dict with this shape:

event = {
    "method":  "POST",               # HTTP method
    "path":    "/webhooks/github",   # URL path
    "headers": {                     # all request headers
        "content-type":   "application/json",
        "x-github-event": "pull_request",
    },
    "query":   {"token": "abc123"},  # URL query parameters
    "body":    {                     # parsed as JSON if possible, else string
        "action": "opened",
        "pull_request": {"title": "..."},
    },
}

The response

Your agent's return value becomes the HTTP response. Return a dict with status, body, and optional headers:

# Full control
return {
    "status":  200,
    "body":    {"result": "processed"},
    "headers": {"x-request-id": request_id},
}

# Error
return {"status": 400, "body": {"error": "Missing required field"}}

# Shorthand: any dict without a "body" key becomes a 200 with that dict as the body
return {"summary": summary}   # → 200 {"summary": "..."}

A complete webhook handler

agents/webhook.star

config(model = "claude-sonnet")

def dict_get(d, key, default = None):
    """Safe dict access — Starlark dicts don't have .get()."""
    if key in d:
        return d[key]
    return default

def agent(event):
    path = event["path"]

    if path == "/ping":
        return {"status": 200, "body": {"pong": True}}

    if path == "/github":
        return handle_github(event)

    if path == "/alert":
        return handle_alert(event)

    return {"status": 404, "body": {"error": "Unknown path: " + path}}

def handle_github(event):
    body   = event["body"]
    action = dict_get(body, "action", "unknown")

    summary = prompt(
        "Summarize this GitHub event and suggest actions:\n" + repr(body),
        max_tokens = 200,
    )
    return {"status": 200, "body": {"action": action, "summary": summary}}

def handle_alert(event):
    body     = event["body"]
    severity = dict_get(body, "severity", "unknown")

    diagnosis = prompt(
        "You are an SRE. Diagnose this alert — root cause, mitigation, escalation:\n" +
        repr(body),
        max_tokens = 300,
    )
    return {"status": 200, "body": {"severity": severity, "diagnosis": diagnosis}}

Testing

# Start the server
chidori serve agents/webhook.star --port 8080

# Built-in health check
curl http://localhost:8080/health

# No-LLM ping
curl http://localhost:8080/ping

# GitHub webhook
curl -X POST http://localhost:8080/github \
  -H "Content-Type: application/json" \
  -d '{"action": "opened", "pull_request": {"title": "Add login", "user": {"login": "alice"}}}'

# Alert
curl -X POST http://localhost:8080/alert \
  -H "Content-Type: application/json" \
  -d '{"severity": "high", "service": "payments", "message": "p99 latency > 2s"}'

Routing patterns

Because agent(event) is just a Starlark function, routing is whatever you write:

def agent(event):
    path   = event["path"]
    method = event["method"]

    # Path-based
    if path == "/webhooks/github":
        return handle_github(event)
    if path == "/webhooks/slack":
        return handle_slack(event)

    # Method-based
    if path == "/tasks" and method == "POST":
        return create_task(event)
    if path == "/tasks" and method == "GET":
        return list_tasks(event)

    # Header-based
    if "x-github-event" in event["headers"]:
        return handle_github(event)

    return {"status": 404, "body": {"error": "Not found"}}

Starlark gotchas for event handlers

Starlark dicts do not have a .get() method. Use in checks or a helper:

# Option 1: explicit check
if "action" in body:
    action = body["action"]
else:
    action = "unknown"

# Option 2: reusable helper
def dict_get(d, key, default = None):
    if key in d:
        return d[key]
    return default

action = dict_get(body, "action", "unknown")

Sessions come for free

Every request handled by chidori serve is a session with an ID, a recorded call log, and a replay endpoint — so any webhook that hit production can be reproduced locally with zero LLM spend. See Checkpoint & Replay.

Was this page helpful?