Develop: Agents

How to author, test, and ship an agent on AcasoOS. For the conceptual model (what agents are, overrides, streaming) see Concepts: Agents; this page is the engineering guide. The in-repo counterpart is docs/agents/authoring.md.

An agent is three things in one Python module: @tool-decorated async functions, an Agent subclass declaring configuration, and a register_agent(...) call at import time. The worked example to copy is packages/agenthub/src/agenthub/examples/team_finder.py.

1. Pick the location

<client_slug> is the tenant slug with dashes replaced by underscores (acme-corp → acme_corp). Client agents never go in examples/ — that directory is reference material (its HR flavor is incidental; AcasoOS is domain-general), and the split is what keeps client work reviewable and upgrade-safe. See packages/agenthub/src/agenthub/custom/README.md and packages/agenthub/src/agenthub/examples/README.md.

Note: Agents are gated by scopes, not hard-isolated by namespace the way bespoke Cultivator Kinds are. On a shared deployment, an agent module that is loaded is runnable by any caller whose scopes match — and acaso_os:admin matches everything. Data stays safe (tools read under the caller's tenant via RLS), but the agent's prompt and behavior are not tenant-private. Ship agents that embed client-confidential logic to that client's standalone deployment, not to the shared one.

2. Write the tools

A tool is an async function decorated with @tool (from agenthub.tools, re-exporting acaso_shared.tools.decorator). The docstring becomes the LLM-visible description; the type hints become the argument schema.

from acaso_shared.tenancy.context import current_tenant_id
from acaso_shared.tenancy.rls import tenant_scope
from datahub.infrastructure.database.connection import get_engine
from sqlalchemy import text

from agenthub.tools import tool


@tool
async def find_people_with_skill(skill_name: str) -> list[dict]:
    """Find every person in the tenant who has the given skill.

    Args:
        skill_name: exact skill name as catalogued in DataHub.
    """
    tenant_id = current_tenant_id()
    if tenant_id is None:
        raise RuntimeError("no tenant in context — tools require a request")
    async with tenant_scope(get_engine(), tenant_id) as conn:
        rows = (await conn.execute(text("SELECT ..."), {"name": skill_name})).mappings().all()
    return [dict(r) for r in rows]

Rules that the runtime relies on:

• tenant_id comes from current_tenant_id() — never from a function parameter. The LLM must not be able to choose the tenant.
• Reads go through tenant_scope(get_engine(), tenant_id) so RLS is bound on the connection.
• Exceptions raised inside a tool are caught by the graph runner and returned to the LLM as {"error": "..."} — raise freely with clear messages; don't swallow.
• @tool(required_scopes=("context:read",)) adds a per-tool gate. The runtime filters gated tools out of the LLM's tool list AND re-checks at invocation time (double lock).
• For memory access, reuse the graphium tools instead of writing your own: search_context, recall_context, wake_up_context, ingest_context from agenthub.tools.graphium. See agenthub.examples.memory_taxonomy for an agent built entirely from them.

3. Declare the agent

from agenthub.agent import Agent, register_agent


class SupportTriager(Agent):
    name = "support-triager"
    version = "1.0"
    model = "anthropic/claude-sonnet-4-6"   # optional; omit to use ACASO_DEFAULT_MODEL
    description = "Triages inbound support tickets against the tenant's product graph."
    system_prompt = (
        "You triage support tickets for {{tenant.org_name}}. "
        "Ground every answer in tool output; never invent ticket data."
    )
    tools = (find_people_with_skill,)
    protocols = ("ag-ui", "a2a", "mcp")
    streaming = True
    session_persistence = "postgres"


register_agent(SupportTriager)

Field reference (packages/agenthub/src/agenthub/agent/base.py):

• namerequired, unique, kebab-case. Registration raises on collision with a different class.
• versionsemver string, default "1.0". Bump on behavior change.
• modelLiteLLM-routable string (anthropic/..., openai/..., openrouter/...). Omit to inherit the platform default (ACASO_DEFAULT_MODEL). Tenants can override per tenant without code.
• system_promptsupports {{tenant.org_name}}, {{tenant.slug}}, {{subject.full_name}}, {{subject.id}}, {{date.today}} placeholders, interpolated at run time. This is only the default: a tenant's production Prompt Version (ADR-0006) shadows it when one exists.
• toolstuple of @tool functions. Per-tenant overrides can disable individual tools by name.
• protocolstuple of the protocol surfaces this agent exposes, default ("ag-ui",). Add "a2a" to turn on the agent-to-agent responder (POST /v1/agents/<name>/a2a, JSON-RPC message/send / message/stream) plus its Agent Card at GET /v1/agents/<name>/.well-known/agent-card.json (consumed by other agents/services; ADR-0019), and/or "mcp" to surface the agent as a tool (agent.<name>) on AgentHub's single Streamable-HTTP MCP endpoint at /mcp (consumed by LLM hosts like Claude Desktop / IDEs; ADR-0020). At least one; any combination — e.g. protocols = ("ag-ui", "a2a", "mcp").
• knowledgelist of DataHub template references (e.g. ["examples/hr_skills"]). In v0 this is surfaced as registry metadata; the template-tool auto-wiring (FR-L2-005) is not active yet, so wire the tools you need explicitly in tools.
• required_scopesauto-derives to ("agents:<name>:run",). Callers whose scopes don't match get a 404 (not 403) so the agent's existence doesn't leak.
• streamingwhether POST /v1/agents/{name}/stream is supported (default true).
• session_persistence"postgres" (LangGraph checkpoints keyed by chat id) or "none" for stateless agents.
• auto_titleset false to opt out of background chat-title generation.
• build_graph(checkpointer)override to author a custom LangGraph workflow; the default is a ReAct loop (agenthub.runtime.graph_builder.build_default_graph). For multi-agent routing, subclass Supervisor — see agenthub.examples.concierge.

4. Register the module with the runtime

AgentHub imports agent modules at startup from the ACASO_AGENT_MODULES env var (comma-separated dotted paths). The code default lists the three example agents — a local-dev convenience so a fresh make setup has something to chat with, not a deployment recommendation:

ACASO_AGENT_MODULES=agenthub.examples.team_finder,agenthub.examples.performance_companion,agenthub.examples.concierge

A client deployment sets the variable explicitly to exactly that client's agents (examples excluded on purpose) and restarts AgentHub:

ACASO_AGENT_MODULES=agenthub.custom.acme_corp.support_triager

An empty string (ACASO_AGENT_MODULES="") loads no agents at all — useful for data-plane-only deployments. A module that fails to import logs the error and is skipped — boot does not crash, so check the startup log when an agent is missing from GET /v1/agents.

5. Grant access

The agent is invisible until a role grants its scope. Add agents:<name>:run to the right role (console → Roles, or PATCH /v1/roles/{slug}). acaso_os:admin matches everything and needs no change. Verify as a non-admin user that the agent appears in GET /v1/agents and that an ungated user gets a 404.

6. Test

Tests live in packages/agenthub/tests/unit/ (AAA pattern mandatory). The runner is testable without Postgres or a live LLM: monkeypatch the checkpointer with LangGraph's MemorySaver and script agenthub.litellm_config.router.acompletion — copy the fixtures in packages/agenthub/tests/unit/orchestrator/conftest.py.

What to cover, minimum:

• Each tool: happy path and the no-tenant-in-context failure, with the DB call stubbed.
• The agent run: scripted LLM responses through the graph, asserting the final message and tool-call sequence.
• Scope behavior: tool filtered out when caller scopes don't match.

uv run pytest packages/agenthub/tests/unit/ -k support_triager

7. Run it locally

make run            # DataHub :8000
make run-agenthub   # AgentHub :8001 (mounts every DataHub router in dev)
make seed           # mock people + skills, if your tools need data

Then chat with it from the console's Agents page, or hit the API directly:

curl -sS -X POST http://localhost:8001/v1/agents/support-triager/run \
  -H "Authorization: Bearer $API_KEY" \
  -H "X-Acaso-User-Id: $USER_ID" \
  -H "Content-Type: application/json" \
  -d '{"message": "Who can take a Kubernetes ticket?"}'

Ship checklist

• Module in the right directory (examples/ vs custom/<client_slug>/), registered at import.
• Tools pull tenant from the ContextVar; no tenant-shaped parameters; reads under tenant_scope.
• required_scopes reviewed; role grants planned for the tenant.
• Unit tests in AAA shape; make lint format_check type_check test green.
• ACASO_AGENT_MODULES updated on every deployment that should load the agent — and examples excluded everywhere outside local dev.
• Confidential-logic check: bespoke prompt/behavior only on the client's own deployment.