# Project export: OnCall This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2026 - Tagline: Talk to your infrastructure. A voice-native ops copilot that remembers your past incidents and proposes the fix that worked last time. - Devpost: https://devpost.com/software/oncall-9e5yl6 - GitHub: https://github.com/BharatKatyal/OnCall/ - Video: https://www.youtube.com/embed/ElI0jXTjVv0?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — BharatM5Pro (3 commits) ## Devpost submission (written by the team) ### Inspiration Incident response is the worst time to type. Your hands are already on the keyboard, you're flipping between dashboards and logs and runbooks, and the clock's running. The thing that would actually help, knowing how you fixed this exact problem last time, is usually stuck in someone's head or a Slack thread you can't find. We wanted infra you could just talk to that remembers what happened before. ### What it does OnCall is a voice copilot for on-call engineers. You ask it out loud what's wrong and it answers back in voice. Ask "what's broken in prod?" and instead of just reading you a number, it'll say something like "gateway DB is near its connection limit. Last time this happened you bumped max_connections to 200 and added a PgBouncer pooler. Want the same fix?" So it does the recall, the diagnosis, and a suggested fix without you touching the keyboard. ### How we built it Voice is Deepgram's Voice Agent API, STT, turn-taking, TTS. We fed it our ops vocab (Lambda, Cognito, PgBouncer, max_connections, ARNs) so it stops mangling jargon. Memory is Redis, used three ways past caching: agent memory for incident history across sessions, vector search over a corpus of runbooks and postmortems, and semantic caching so repeat questions skip the LLM call. Running on Redis Cloud. The agent itself we built with Claude Code. It calls a few ops tools (log query, metric lookup, fix proposal) against a harness we seeded with real incident data, including an actual Postgres connection-exhaustion outage we dealt with. Flow: voice → agent → Redis → voice. Diagram below. ### Challenges we ran into Voice latency was the big one, getting it fast enough that it doesn't feel like a walkie-talkie. STT kept choking on ops terms until we tuned it. And figuring out a memory schema that separates "what's happening right now this session" from "what we've learned over time" took a few tries. ### Accomplishments we're proud of The voice actually matters here, it's not a button we bolted on, you genuinely can't type this fast mid-incident. And the memory recall works. Watching it pull up a specific past outage and suggest the fix that worked got a real reaction from people who tried it. ### What we learned Redis past caching is a bigger deal than we expected, adding memory and vector search turned a stateless bot into something that actually accumulates knowledge. And building for voice forced us to be way more disciplined. A bot that talks out loud can't hide behind a wall of text, it has to actually know the answer. ### What's next Wiring up live AWS (CloudWatch, Lambda, RDS) behind the same interface, letting it actually run fixes with a human approving each step, and shared team memory so the whole rotation inherits one brain instead of everyone learning the same lessons separately. ## README (from the GitHub repository) # OnCall ## Detected evidence (automated analysis) Indexed codebase: 13 recognized source files, 74 KB. - Python (language) — detected in the code - Redis (technology) — detected in the code ## Codebase structure (from repository index) ### Files (18 of 18) ``` .gitignore k8/k8Setup.md k8/oncall-rbac.yaml oncall.py README.md voice-agent-function-calling/.env.example voice-agent-function-calling/.gitignore voice-agent-function-calling/business_logic.py voice-agent-function-calling/client.py voice-agent-function-calling/config.py voice-agent-function-calling/functions.py voice-agent-function-calling/k8s.py voice-agent-function-calling/LICENSE voice-agent-function-calling/preflight.py voice-agent-function-calling/README.md voice-agent-function-calling/redis_store.py voice-agent-function-calling/REDIS.md voice-agent-function-calling/requirements.txt ``` ### Dependencies - voice-agent-function-calling/requirements.txt: certifi, janus@==1.0.0, model2vec, numpy, PyAudio@==0.2.14, python-dotenv, redis@==8.0.0, websockets@==12.0 ### Recent commits (newest first) - added support for real time kubernetes cluster read only - inital - first commit ## Key source files (fetched from GitHub, selected and truncated for size) ### k8/k8Setup.md ```markdown Apply the oncall-rbac.yaml Part 1 — Build a read-only kubeconfig (5 min) Run these to mint a token and assemble a standalone kubeconfig the agent will use: # Get your cluster's server URL and CA from your current (admin) context CLUSTER_NAME=$(kubectl config view --minify -o jsonpath='{.clusters[0].name}') SERVER=$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}') kubectl config view --minify --raw -o jsonpath='{.clusters[0].cluster.certificate-authority-data}' | base64 -d > /tmp/oncall-ca.crt # Mint a short-lived read-only token TOKEN=$(kubectl create token oncall-readonly -n default --duration=8h) # Assemble a dedicated kubeconfig kubectl --kubeconfig=/tmp/oncall-kubeconfig config set-cluster "$CLUSTER_NAME" \ --server="$SERVER" --certificate-authority=/tmp/oncall-ca.crt --embed-certs=true kubectl --kubeconfig=/tmp/oncall-kubeconfig config set-credentials oncall-readonly --token="$TOKEN" kubectl --kubeconfig=/tmp/oncall-kubeconfig config set-context oncall \ --cluster="$CLUSTER_NAME" --user=oncall-readonly kubectl --kubeconfig=/tmp/oncall-kubeconfig config use-context oncall Verify the lock works before wiring anything — this is the gate: # Should SUCCEED (read): kubectl --kubeconfig=/tmp/oncall-kubeconfig get pods -A # Should be DENIED (write) — proving RBAC holds: kubectl --kubeconfig=/tmp/oncall-kubeconfig delete ns default --dry-run=server ``` ### voice-agent-function-calling/REDIS.md ```markdown # OnCall — Redis incident-memory plane `search_incident_memory` is backed by **Redis Stack** (RediSearch vector index) + local **model2vec** embeddings (`redis_store.py`). If Redis is unreachable, the tool falls back to the verbatim INC-0412 record so the demo never breaks. Redis is used three ways (matching the DevPost): - **Agent memory** — incidents persist as hashes under `incident:` across sessions. - **Vector search** — cosine KNN over the incident corpus (`incidents_idx`). - **Semantic cache** — a second vector index (`qcache_idx`) over past queries; a new query within `CACHE_SIM_THRESHOLD` (0.50) cosine of a prior one returns the cached result instantly. Console prints `CACHE HIT` / `cache miss` for the demo. To reset the cache between demo runs: ```bash docker exec oncall-redis redis-cli --scan --pattern 'qcache:*' | xargs -r docker exec oncall-redis redis-cli DEL ``` ## Local Redis (Docker) — current setup ```bash docker run -d --name oncall-redis -p 6380:6379 -p 8001:8001 redis/redis-stack:latest ``` > Host port is **6380** (not 6379) because another project's `redis:7-alpine` > already holds 6379. `redis_store.py` defaults to `localhost:6380`. - Redis: `localhost:6380` - RedisInsight UI: http://localhost:8001 - Module check: `docker exec oncall-redis redis-cli MODULE LIST | grep search` Container lifecycle: ```bash docker start oncall-redis # after a reboot docker stop oncall-redis docker rm -f oncall-redis # remove ``` ## Seed the corpus Auto-seeds on first search, or explicitly: ```bash ./venv/bin/python redis_store.py # seeds + runs a sample query ``` ## Redis Cloud (ACTIVE) The app is pointed at Redis Cloud via a single `REDIS_URL` in the app-local `.env` (gitignored). `client.py` / `redis_store.py` auto-load `.env` (python-dotenv), so no `source` is needed. ``` # .env (not committed) REDIS_URL=redis://default:@: ``` `get_client()` prefers `REDIS_URL`; if it's unset it falls back to `REDIS_HOST`/`REDIS_PORT`/`REDIS_PASSWORD` (local Docker on 6380). To re-seed the cloud index: `./venv/bin/python redis_store.py`. Redis Cloud includes RediSearch, so the vector index works as-is. The local Docker container is now optional. ``` ### voice-agent-function-calling/requirements.txt ``` PyAudio==0.2.14 websockets==12.0 janus==1.0.0 redis==8.0.0 numpy model2vec certifi python-dotenv ``` ### oncall.py ```python import json import anthropic client = anthropic.Anthropic() SYSTEM_PROMPT = """You are OnCall, a voice-native ops copilot for on-call engineers. The user talks to you out loud during incidents; your replies are read aloud by TTS, so keep them SHORT, spoken-natural, and free of markdown, lists, or symbols. No more than 2-3 sentences per turn. Your job each turn: 1. Understand what the engineer is asking about their infrastructure. 2. Use your tools to investigate: check logs, pull metrics, and ALWAYS search incident memory for relevant past incidents before answering. 3. If you find a relevant past incident, lead with it: name what happened last time and what fixed it. This recall is your most valuable behavior. 4. When asked to act, call propose_fix and read the proposed remediation back conversationally. Never claim you executed anything — you propose, the human approves. Voice rules: - Speak like a calm, terse senior SRE. No filler, no "great question." - Spell infra terms naturally ("max connections", "pee-gee-bouncer"). - If a tool returns nothing, say so plainly and suggest the next check. Always call search_incident_memory before giving any diagnosis.""" TOOLS = [ { "name": "query_logs", "description": "Query recent log entries for a service. Returns error and warning lines from the last N minutes.", "input_schema": { "type": "object", "properties": { "service": {"type": "string", "description": "Service name, e.g. 'gateway-logdb'"}, "minutes": {"type": "integer", "description": "Lookback window in minutes"} }, "required": ["service"] } }, { "name": "get_metric", "description": "Get the current value and recent trend for an infrastructure metric.", "input_schema": { "type": "object", "properties": { "metric": {"type": "string", "description": "e.g. 'db_connections', 'disk_usage', 'cpu'"}, "resource": {"type": "string", "description": "Resource identifier"} }, "required": ["metric", "resource"] } }, { "name": "search_incident_memory", "description": "Semantic search over past incidents and runbooks in Redis. Returns the most relevant past incident with its resolution. Call this before every diagnosis.", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "Natural-language description of the current symptom"} }, "required": ["query"] } }, { "name": "propose_fix", "description": "Generate a proposed remediation for a diagnosed issue. Returns the steps; does NOT execute them.", "input_schema": { "type": "object", "properties": { "issue": {"type": "string", "description": "The diagnosed problem"}, "based_on_incident": {"type": "string", "description": "ID of the past incident this fix is modeled on, if any"} }, "required": ["issue"] } } ] # --- Mock implementations (real CNPG numbers baked in) --- def query_logs(service, minutes=15): if "logdb" in service or "gateway" in service: return { "service": service, "errors": [ "FATAL: sorry, too many clients already", "remaining connection slots reserved for superuser", ], "warning_count": 47, } return {"service": service, "errors": [], "warning_count": 0} def get_metric(metric, resource): if metric == "db_connections": return {"metric": metric, "resource": resource, "current": 98, "limit": 100, "trend": "rising", "unit": "connections"} if metric == "disk_usage": return {"metric": metric, "resource": resource, "current": 9.4, "limit": 10, "trend": "rising", "unit": "Gi"} return {"metric": metric, "resource": resource, "current": None} def search_incident_memory(query): return { "match": { "id": "INC-0412", "title": "replyagent-gateway-logdb connection exhaustion", "symptom": "Postgres hit max_connections, app threw 'too many clients'", "resolution": "Raised max_connections to 200, deployed PgBouncer Pooler " "to multiplex connections, expanded PVC 10Gi to 20Gi after " "disk-full CrashLoopBackOff.", "resolved_in_minutes": 38, }, "similarity": 0.91, } def propose_fix(issue, based_on_incident=None): return { "issue": issue, "modeled_on": based_on_incident, "steps": [ "Patch the CNPG cluster: set max_connections to 200.", "Deploy a PgBouncer Pooler to multiplex connections.", "Pre-emptively expand the PVC from 10Gi to 20Gi to avoid disk-full crashloop.", ], "requires_human_approval": True, } DISPATCH = { "query_logs": query_logs, "get_metric": get_metric, "search_incident_memory": search_incident_memory, "propose_fix": propose_fix, } def run_turn(user_text, history): history.append({"role": "user", "content": user_text}) while True: resp = client.messages.create( model="claude-opus-4-8", max_tokens=1024, system=SYSTEM_PROMPT, tools=TOOLS, messages=history, ) history.append({"role": "assistant", "content": resp.content}) if resp.stop_reason != "tool_use": text = "".join(b.text for b in resp.content if b.type == "text") return text, history results = [] for block in resp.content: if block.type == "tool_use": print(f" [tool] {block.name}({block.input})") [truncated — 599 more characters] ``` ### voice-agent-function-calling/config.py ```python ARTIFICIAL_DELAY = { "database": 0.0, "external_api": 0.0, # Not in use in this reference implementation but left as an example for simulating different delays "heavy_computation": 0.0 # Not in use in this reference implementation but left as an example for simulating different delays } # Mock data settings MOCK_DATA_SIZE = { "customers": 1000, "appointments": 500, "orders": 2000 } # Database settings (if using SQLite) # Not in use in this reference implementation but left as an example for how to potentially integrate with a DB DATABASE_CONFIG = { "path": "business_data.db", "enable": False # Set to True to use actual SQLite instead of mock data } ``` ### k8/oncall-rbac.yaml ```yaml apiVersion: v1 kind: ServiceAccount metadata: name: oncall-readonly namespace: default --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: oncall-readonly-view subjects: - kind: ServiceAccount name: oncall-readonly namespace: default roleRef: kind: ClusterRole name: view # built-in read-only, excludes Secrets apiGroup: rbac.authorization.k8s.io --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: oncall-readonly-extra rules: - apiGroups: [""] resources: ["nodes", "namespaces"] verbs: ["get", "list", "watch"] - apiGroups: ["metrics.k8s.io"] resources: ["*"] verbs: ["get", "list"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: oncall-readonly-extra subjects: - kind: ServiceAccount name: oncall-readonly namespace: default roleRef: kind: ClusterRole name: oncall-readonly-extra apiGroup: rbac.authorization.k8s.io ``` ### voice-agent-function-calling/preflight.py ```python """No-microphone preflight: connects with the REAL settings from client.py, sends them, and prints the server's first replies. Confirms the schema, the Anthropic provider + key, the function definitions, and that the greeting plays — all without talking. Run this before client.py to catch config errors fast. python preflight.py """ import asyncio import json import os import ssl import certifi import websockets import client # reuse the exact SETTINGS / URL the real app uses async def main(): dg = os.environ.get("DEEPGRAM_API_KEY") an = os.environ.get("ANTHROPIC_API_KEY") if not dg: print("FAIL: DEEPGRAM_API_KEY not set"); return if not an: print("FAIL: ANTHROPIC_API_KEY not set"); return settings = client.SETTINGS.copy() settings["agent"]["think"]["endpoint"]["headers"]["x-api-key"] = an ssl_context = ssl.create_default_context(cafile=certifi.where()) print(f"Connecting to {client.VOICE_AGENT_URL} ...") async with websockets.connect( client.VOICE_AGENT_URL, ssl=ssl_context, extra_headers={"Authorization": f"Token {dg}"}, ) as ws: await ws.send(json.dumps(settings)) applied = False try: for _ in range(8): msg = await asyncio.wait_for(ws.recv(), timeout=8) if isinstance(msg, bytes): print(f"