7.1 Blueprint A · SaaS In-Product Assistant
⚡ Runnable end-to-end
Outcome — FastAPI app with SSE streaming + per-tenant agent pool + permission rules; chat from the browser, agent calls scoped business tools. Stack — Python · FastAPI · SSE · custom Tools · PermissionEngine · session pool with TTL. Source — examples/saas-assistant/Run — uv sync && uv run uvicorn app.main:app --reload
Scenario: you run a project-management SaaS. Users want "help me schedule this project plan" or "summarize last week's tasks" inside the product. You want to embed Agentao as the brain, expose only the tools it needs, and stream responses to the browser.
Who & why
- Product shape: existing web app (frontend + backend API)
- Users: logged-in tenants, each with scoped data
- Pain point: users write their own prompts in scratch copies of ChatGPT and paste results back — data leaks, no in-product integration
Architecture
Browser (React)
│
│ POST /chat { session_id, message } ── SSE stream back
▼
FastAPI backend
├─ Auth middleware (tenant_id, user_id)
├─ Session pool (see 6.7)
│ │
│ ▼
└─ Agentao instance per (tenant_id, session_id)
├─ working_directory = /data/tenants/{tenant_id}/{session_id}
├─ Custom tools: list_projects, create_task, assign_user
├─ PermissionEngine: WORKSPACE_WRITE, deny rules for the built-in writers
└─ agent.astream() → SSE bridgeKey code
1 · Custom tools that hit your backend
python
# tools/project_tools.py
from agentao.host import Tool
import httpx
class ListProjectsTool(Tool):
def __init__(self, tenant_id: str, api_client: httpx.Client):
self._tenant_id = tenant_id
self._api = api_client
@property
def name(self): return "list_projects"
@property
def description(self): return "List projects visible to the current user"
@property
def parameters(self):
return {"type": "object", "properties": {
"status": {"type": "string", "enum": ["active", "archived", "all"]}
}}
@property
def is_read_only(self): return True
def execute(self, status: str = "active") -> str:
r = self._api.get(f"/api/v1/projects",
params={"tenant_id": self._tenant_id, "status": status},
timeout=10)
r.raise_for_status()
return r.textpython
class CreateTaskTool(Tool):
def __init__(self, tenant_id: str, api_client: httpx.Client):
self._tenant_id = tenant_id
self._api = api_client
@property
def name(self): return "create_task"
@property
def description(self): return "Create a task in a project"
@property
def parameters(self):
return {"type": "object", "required": ["project_id", "title"], "properties": {
"project_id": {"type": "string"},
"title": {"type": "string"},
"assignee_email": {"type": "string"},
"due_date": {"type": "string", "format": "date"},
}}
@property
def requires_confirmation(self): return True # write action
def execute(self, **kwargs) -> str:
r = self._api.post(f"/api/v1/tasks",
json={"tenant_id": self._tenant_id, **kwargs},
timeout=10)
r.raise_for_status()
return r.text2 · FastAPI endpoint with SSE streaming
python
# app.py
from contextlib import aclosing
from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse
from agentao.host import TextDelta
from pathlib import Path
import httpx, json
from .tools.project_tools import ListProjectsTool, CreateTaskTool
from .auth import current_user
from .pool import get_or_create_agent
app = FastAPI()
api_client = httpx.Client(base_url="http://internal-api")
@app.post("/chat")
async def chat(payload: dict, user=Depends(current_user)):
session_id = payload["session_id"]
message = payload["message"]
workdir = Path(f"/data/tenants/{user.tenant_id}/{session_id}")
workdir.mkdir(parents=True, exist_ok=True)
agent, lock = await get_or_create_agent(
session_id=f"{user.tenant_id}:{session_id}",
workdir=workdir,
tenant_id=user.tenant_id,
)
async def sse():
async with lock: # one turn per session; without it a second request gets TurnInProgressError
# astream() subscribes to the agent's transport; never swap agent.transport per request.
# aclosing cancels the turn if the client goes away.
async with aclosing(agent.astream(message)) as stream:
async for item in stream:
if isinstance(item, TextDelta):
frame = {"type": "llm_text", "chunk": item.text}
yield f"data: {json.dumps(frame)}\n\n"
else: # TurnOutcome: the answer (the chunks also carry narration before tool calls)
done = {"text": item.text, "is_answer": item.is_answer}
yield f"event: done\ndata: {json.dumps(done)}\n\n"
return StreamingResponse(sse(), media_type="text/event-stream")3 · Pool wiring
python
# pool.py
import asyncio
from agentao import Agentao
from agentao.permissions import PermissionEngine, PermissionMode
from .tools.project_tools import ListProjectsTool, CreateTaskTool
_pool: dict[str, tuple[Agentao, asyncio.Lock]] = {}
async def get_or_create_agent(session_id, workdir, tenant_id):
existing = _pool.get(session_id)
if existing:
return existing # same agent and transport across requests
# Deny the built-in writers by rule; read-only mode would deny
# CreateTaskTool too. create_task keeps asking via requires_confirmation.
engine = PermissionEngine(project_root=workdir, rules=[
{"tool": "write_file", "action": "deny"},
{"tool": "replace", "action": "deny"},
{"tool": "run_shell_command", "action": "deny"},
{"tool": "save_memory", "action": "deny"},
])
engine.set_mode(PermissionMode.WORKSPACE_WRITE)
agent = Agentao(
working_directory=workdir,
permission_engine=engine,
)
agent.tools.register(ListProjectsTool(tenant_id, api_client))
agent.tools.register(CreateTaskTool(tenant_id, api_client))
_pool[session_id] = (agent, asyncio.Lock())
return _pool[session_id]Use the full AgentPool from 6.7 for TTL + LRU eviction.
Frontend skeleton
ts
const es = new EventSource(`/chat`, { method: "POST", body: JSON.stringify({session_id, message}) });
es.onmessage = (e) => {
const ev = JSON.parse(e.data);
if (ev.type === "llm_text") append(ev.chunk);
// Tool activity is not in this stream: relay agent.events() (ToolLifecycleEvent)
// on a second channel. Confirmations are answered by the transport the agent was
// built with, e.g. SdkTransport(confirm_tool=...) — see 4.5.
};
es.addEventListener("done", (e) => {
const done = JSON.parse(e.data);
if (done.is_answer) finalize(done.text); // the answer
else showError(done.text); // e.g. an [LLM API error: …] text
es.close();
});⚠️ Pitfalls
Day-2 bugs from real SaaS deployments
Each row below is a real production incident. Skim them before you ship — the fixes are cheap now and expensive later.
| Day-2 bug | Root cause | Fix |
|---|---|---|
| Cross-tenant data leak | Tool captured tenant_id at construction time, but pool reused an agent for another tenant | One agent per (tenant_id, session_id) — never share across tenants |
| "My tasks vanished!" | clear_history() called when the SDK reset, but backend memory DB retained user-wide notes and bled into another session | Use project-scope memory only; if you mount user-scope, key it by tenant_id+user_id |
| Agent "hangs" forever | No per-chat timeout | Wrap chat() in asyncio.wait_for (6.7 Control 3) |
| SSE stream stops mid-reply | Frontend reconnect buffers didn't match server; browser idle kill | Send periodic : keep-alive\n\n every 15s |
| Confirm modal blocks forever | confirm_tool ran on the event-loop thread | See 4.5 Web modal — use asyncio.run_coroutine_threadsafe |
Runnable code
The full project lives in-repo at examples/saas-assistant/ — see the top-of-page "Run this example" link.
bash
cd examples/saas-assistant
uv sync && uv run uvicorn app.main:app --reload