Compute
Build a container that listens for webhooks, spawns coding agents, and tracks state in SQLite.
A single long-running container that:
Volume.This pattern is useful when you want a lightweight orchestration layer that turns external events into autonomous coding sessions without running a full queue system.
Container secrets work like environment variables, but are managed through Chalk so they never appear in your source code or image layers. Declare the keys you need upfront — the container won’t start if any required secret is missing.
from chalkcompute import Container, Image, Secret, Volume
container = Container(
image=image,
secrets=[
Secret(name="OPENAI_API_KEY"),
Secret(name="GITHUB_WEBHOOK_SECRET"),
Secret(name="CONTROLLER_URL"),
Secret(name="CONTROLLER_API_KEY"),
],
# ...
)Each Secret(name="...") references a secret stored in your Chalk project settings.
Secrets are named after the environment variable they inject — use the as_env_var
parameter if you need to alias a secret to a different environment variable name.
webhook-orchestrator/
├── orchestrator.py # FastAPI app: webhook handler + agent tracker
└── deploy.py # Chalk container definition
We’ll walk through each file, then deploy the whole thing with a single command.
The orchestrator is a small FastAPI app. It does three things: initializes a SQLite database, accepts webhook POST requests, and calls the Chalk Controller to spawn agent sessions.
We store agent state in SQLite on a mounted volume. This means the data survives container restarts — if the orchestrator crashes at 3 AM, it comes back up and knows exactly which agents are still running.
# orchestrator.py
import sqlite3
import os
import httpx
from datetime import datetime, timezone
from contextlib import contextmanager
from fastapi import FastAPI, Request, HTTPException
DB_PATH = os.environ.get("DB_PATH", "/data/agents.db")
CONTROLLER_URL = os.environ["CONTROLLER_URL"]
CONTROLLER_API_KEY = os.environ.get("CONTROLLER_API_KEY", "")
app = FastAPI()
def init_db() -> None:
with _db() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS agents (
id TEXT PRIMARY KEY,
webhook_event TEXT NOT NULL,
repo TEXT NOT NULL,
prompt TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
session_id TEXT,
task_id TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
@contextmanager
def _db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()Nothing exotic here — a single agents table that tracks the mapping from webhook
event to controller session.
When a webhook arrives, we parse it, insert a row, and kick off an agent session
in the background. The endpoint returns 200 immediately so the webhook source
doesn’t time out.
import uuid
import asyncio
@app.on_event("startup")
def on_startup() -> None:
init_db()
@app.post("/webhook")
async def handle_webhook(request: Request) -> dict:
payload = await request.json()
# Extract the relevant fields — adapt this to your webhook source.
repo = payload.get("repository", {}).get("full_name", "unknown/unknown")
event_type = request.headers.get("X-GitHub-Event", "unknown")
prompt = _build_prompt(event_type, payload)
agent_id = str(uuid.uuid4())
now = datetime.now(timezone.utc).isoformat()
with _db() as conn:
conn.execute(
"INSERT INTO agents (id, webhook_event, repo, prompt, status, created_at, updated_at)"
" VALUES (?, ?, ?, ?, 'pending', ?, ?)",
(agent_id, event_type, repo, prompt, now, now),
)
# Fire-and-forget: spawn the agent without blocking the response.
asyncio.create_task(_spawn_agent(agent_id, repo, prompt))
return {"agent_id": agent_id, "status": "pending"}Different webhook events need different instructions. A new issue gets an investigative prompt; a PR review comment gets a targeted fix prompt.
def _build_prompt(event_type: str, payload: dict) -> str:
if event_type == "issues" and payload.get("action") == "opened":
issue = payload["issue"]
return (
f"Investigate GitHub issue #{issue['number']}: {issue['title']}\n\n"
f"{issue.get('body', 'No description provided.')}\n\n"
"Diagnose the root cause and open a PR with a fix."
)
if event_type == "issue_comment":
comment = payload["comment"]
issue = payload["issue"]
return (
f"Respond to comment on #{issue['number']}: {issue['title']}\n\n"
f"Comment by @{comment['user']['login']}:\n{comment['body']}\n\n"
"Address the feedback and push an updated commit."
)
return f"Handle {event_type} event for this repository."This is where the orchestrator calls the Chalk Controller. It creates a session (which provisions a workspace), then submits a task with the generated prompt. As each step completes, it updates the SQLite row.
async def _spawn_agent(agent_id: str, repo: str, prompt: str) -> None:
try:
async with httpx.AsyncClient(
base_url=CONTROLLER_URL,
headers={"Authorization": f"Bearer {CONTROLLER_API_KEY}"},
timeout=120,
) as client:
# Step 1: Create a session (provisions a workspace with the repo).
resp = await client.post("/api/v1/sessions", json={
"description": f"Webhook agent for {repo}",
"repo": repo,
})
resp.raise_for_status()
session = resp.json()
session_id = session["id"]
_update_agent(agent_id, status="session_created", session_id=session_id)
# Step 2: Submit the task.
resp = await client.post(f"/api/v1/sessions/{session_id}/tasks", json={
"prompt": prompt,
"model": "claude-sonnet-4-20250514",
})
resp.raise_for_status()
task = resp.json()
_update_agent(agent_id, status="running", task_id=task["id"])
except Exception as e:
_update_agent(agent_id, status="failed")
raise
def _update_agent(agent_id: str, status: str, session_id: str | None = None, task_id: str | None = None) -> None:
now = datetime.now(timezone.utc).isoformat()
with _db() as conn:
conn.execute(
"UPDATE agents SET status = ?, session_id = COALESCE(?, session_id),"
" task_id = COALESCE(?, task_id), updated_at = ? WHERE id = ?",
(status, session_id, task_id, now, agent_id),
)A simple GET endpoint lets you check on all agents — useful for dashboards or debugging.
@app.get("/agents")
def list_agents(status: str | None = None) -> list[dict]:
with _db() as conn:
if status:
rows = conn.execute(
"SELECT * FROM agents WHERE status = ? ORDER BY created_at DESC", (status,)
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM agents ORDER BY created_at DESC"
).fetchall()
return [dict(row) for row in rows]
@app.get("/health")
def health() -> dict:
return {"status": "ok"}Now we wire everything together with a Container that mounts a Volume for the
database, injects secrets, and starts the FastAPI server.
# deploy.py
from chalkcompute import Container, Image, Secret, Volume
vol = Volume(name="orchestrator-data")
image = (
Image.base("python:3.12-slim")
.pip_install(["fastapi", "uvicorn", "httpx"])
.add_local_file("orchestrator.py", "/app/orchestrator.py")
)
container = Container(
image=image,
name="webhook-orchestrator",
port=8000,
secrets=[
Secret(name="OPENAI_API_KEY"),
Secret(name="GITHUB_WEBHOOK_SECRET"),
Secret(name="CONTROLLER_URL"),
Secret(name="CONTROLLER_API_KEY"),
],
volumes={"orchestrator-data": "/data"},
entrypoint=[
"uvicorn", "orchestrator:app",
"--host", "0.0.0.0",
"--port", "8000",
"--app-dir", "/app",
],
).run()
print(f"Orchestrator URL: {container.info.web_url}")
print(f"Point your GitHub webhook to: {container.info.web_url}/webhook")chalk compute deploy deploy.py
# ✓ Container created successfully
# Container ID: d81f42e9-6ab7-4c3d-9e15-2f0a8b73c4d1
# Name: webhook-orchestrator
# Status: Running
# Pod Name: chalk-container-webhook-orchestrator
# URL: https://d81f42e9-6ab7-4c3d-9e15-2f0a8b73c4d1.compute.chalk.aiCopy the container URL with /webhook appended into your GitHub repository’s webhook settings
(Settings → Webhooks → Add webhook). Select the events you want to trigger agents
for — typically Issues and Issue comments.
GitHub/Linear/Slack Chalk Compute
─────────────────── ─────────────
webhook POST ──────────────▸ ┌───────────────────┐
│ Orchestrator │
│ (FastAPI) │
│ │
│ SQLite on Volume │──▸ /data/agents.db
└────────┬──────────┘
│
│ POST /api/v1/sessions
│ POST /api/v1/sessions/:id/tasks
▼
┌──────────────────┐
│ Chalk Controller │
│ │
│ Provisions a │
│ workspace, │
│ runs the agent │
└──────────────────┘
Each webhook creates one row in SQLite and one agent session in the controller. The volume ensures the database survives restarts, and secrets keep credentials out of your code.