Shipping a Production-Ready Brain+Hands Support Agent (WordPress + Python API)

This build delivers a support/billing agent you can actually ship. It follows Brain+Hands separation, explicit tool contracts, a deterministic state machine, and secure backend tools. Stack: WordPress (frontend), Python FastAPI (tools + orchestrator), Redis (state/cache), Postgres (KB + logs), vector DB (memory), OpenAI/Groq/Anthropic (LLM).

1) Architecture overview
– Brain (LLM policy): Plans, chooses tools, reasons. No direct DB/API access.
– Hands (tools): Idempotent HTTP endpoints with strict schemas. Observable, rate-limited, auditable.
– Orchestrator: State machine controlling turns, tool calls, retries, and timeouts.
– Memory:
– Short-term: per-session scratchpad (Redis).
– Knowledge: vector search over product docs/FAQ.
– Facts: authoritative store lookups (billing, orders).
– Guardrails: auth, PII redaction, tool allowlist, cost/time budgets, circuit breakers.
– Integration: WordPress plugin sends chat events to orchestrator; streaming tokens back to UI.

2) Contracts first (make tools boring and safe)
Define JSON schemas for every tool. Keep them narrow, idempotent, and testable.

Example tool manifest (slice):
{
“name”: “get_order_status”,
“description”: “Return current order state for a given order_id.”,
“method”: “POST”,
“url”: “https://api.example.com/tools/get_order_status”,
“input_schema”: {
“type”: “object”,
“properties”: {
“order_id”: {“type”: “string”, “pattern”: “^[A-Z0-9-]{6,}$”}
},
“required”: [“order_id”],
“additionalProperties”: false
},
“output_schema”: {
“type”: “object”,
“properties”: {
“status”: {“type”: “string”},
“updated_at”: {“type”: “string”, “format”: “date-time”}
},
“required”: [“status”]
},
“timeouts_ms”: 2500,
“retries”: 1
}

3) Hands: secure Python FastAPI tools
– Enforce schema at the edge.
– Require JWT with narrow scopes.
– Add rate limits and audit logs.

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
import time

app = FastAPI(title=”SupportAgentTools”)

class OrderReq(BaseModel):
order_id: str = Field(min_length=6, pattern=r”^[A-Z0-9-]{6,}$”)

class OrderResp(BaseModel):
status: str
updated_at: str | None = None

def auth(scope: str):
def _auth(token=Depends(…)): # your JWT dependency
if scope not in token.scopes:
raise HTTPException(403, “forbidden”)
return token.sub
return _auth

@app.post(“/tools/get_order_status”, response_model=OrderResp)
def get_order_status(req: OrderReq, _=Depends(auth(“order:read”))):
start = time.time()
# query read replica; maintain SLO use respective tools.
– General product usage -> retrieve_docs.
– Anything else -> answer concisely or ask a clarifying question.

6) Orchestrator: a small, reliable state machine
States:
– RECEIVE -> PLAN -> EXECUTE_TOOL? -> OBSERVE -> RESPOND -> END

Pseudo:
def handle_turn(msg, session_id):
budget = Budget(tokens=3000, tools=3, wall_ms=8000)
state = “PLAN”
memory = load_session(session_id)
while budget.ok() and state != “END”:
if state == “PLAN”:
action = llm_policy(memory, tool_manifest)
if action.type == “tool”:
state = “EXECUTE_TOOL”
else:
state = “RESPOND”
elif state == “EXECUTE_TOOL”:
result = call_tool(action.name, action.args, timeout=manifest[action.name].timeouts_ms)
record_observation(result)
state = “OBSERVE”
elif state == “OBSERVE”:
memory.update_with_observation(result)
if need_more_tools(result): state = “PLAN”
else: state = “RESPOND”
elif state == “RESPOND”:
reply = llm_response(memory)
emit_stream(reply)
state = “END”

Controls:
– Max 2 tool calls/turn for latency.
– Tool circuit breaker after 2x p95 failures.
– Token + time budgets enforced per turn.

7) Retrieval that doesn’t hallucinate
– Chunk docs to 300–500 tokens with overlap 50–100.
– Store title, URL, product tags, and last_updated.
– Rerank top 20 -> 5 with a fast cross-encoder or LLM-judge at small context.
– In answers, include “According to