The SDK

The SDK is one Python package, agentgraft. It exposes seven decorators (the public DSL), a small settings dict (the host app config), and one Django management command (the sync). Everything else is internals you can ignore.

#Installation

pip install agentgraft[django]   # Django app: management command, webhook view, template tag
pip install agentgraft           # core only: registry, decorators, schema — no Django required

Python 3.14+ is required. The core package has no runtime dependencies beyond the standard library. Django is an optional extra — install agentgraft[django] for the full Django integration, or agentgraft alone when using the core SDK from another framework (FastAPI, Flask, raw WSGI).

#@ag.user_loader

Resolves an external user id back to your domain user object. Called once per chat turn before any tools run; whatever it returns becomes the first positional argument to every tool function in the same profile.

@ag.user_loader
def load_user(external_id: str) -> User:
    return User.objects.get(pk=external_id)

If your widget is rendered for anonymous visitors too, declare the loader with an is_anonymous keyword and the SDK will pass it on every call. It's True whenever the external id is in the session: namespace the template tag gives logged-out visitors. Loaders that don't declare the keyword still work — the SDK introspects the signature and only forwards what the function expects.

@ag.user_loader(profile="public")
def load_guest(external_id: str, *, is_anonymous: bool = True) -> dict:
    return {"session_key": external_id.removeprefix("session:")}

#@ag.tool

Marks a function as a tool the LLM can call. The first argument is the resolved user object — it's not visible to the LLM, the SDK injects it. Every other argument's type hint becomes a JSON schema field; the docstring becomes the tool description.

@ag.tool
def search_assets(user: User, query: str, limit: int = 10) -> list[dict]:
    """Search the user's assets by name or ticker.

    Use this for stock lookups, crypto symbols, anything in the
    portfolio. Returns up to ``limit`` matches with ticker, name,
    and current quote.
    """
    matches = Asset.objects.filter(owner=user, name__icontains=query)[:limit]
    return [{"ticker": a.ticker, "name": a.name, "price": a.price} for a in matches]

The supported type hints are str, int, float, bool, list[T], dict, Optional[T], Literal[...], Enum subclasses, and Pydantic (v2) models or lists of them. Default values make a parameter optional in the schema. Anything more exotic — dataclasses, other classes — should be unwrapped to primitives at the function boundary.

A Pydantic-typed parameter is sent to the LLM as the model's JSON schema, with references inlined, and the argument reaches your function as a validated instance (None for an Optional one sent as null). One that doesn't validate is a 400 saying where and what is wrong, so the model can correct its call. Use it for structured input that a few primitives can't express:

from pydantic import BaseModel, Field


class LineItem(BaseModel):
    sku: str = Field(description="Product SKU, as returned by search_products.")
    quantity: int = Field(default=1, ge=1, le=99)


@ag.tool
def create_order(user: User, items: list[LineItem], note: str = "") -> dict:
    """Place an order for the user's items."""
    order = Order.objects.create(owner=user, note=note)
    for item in items:  # validated LineItem instances
        order.lines.create(sku=item.sku, quantity=item.quantity)
    return {"id": order.pk}

AgentGraft retries a call once if your webhook times out or returns 502/503/504. Tools that write should call ag.current_request_id() and use it as an idempotency key; the id is the same on the retry. See the Wire protocol page for an example.

#Parameter descriptions

The LLM fills in tool arguments using the parameter name, its type, and — if provided — a natural-language description. Descriptions flow directly into the JSON schema the LLM receives and are one of the highest-leverage ways to improve tool call accuracy. There are two ways to add them; both produce the same result.

#Option 1 — Google-style Args: section (recommended)

Write a standard Args: block in the docstring. The SDK extracts it automatically at sync time — no extra imports, no annotation changes needed. The Args: section is stripped from the tool-level description before it reaches the LLM, so the model sees a clean summary in the tool description and targeted guidance in each parameter's schema field.

@ag.tool
def get_order(user, order_id: str, include_line_items: bool = False) -> dict:
    """Fetch a single order by ID.

    Args:
        order_id: UUID of the order to fetch, as returned by list_orders.
        include_line_items: Pass True to include individual line items in
            the response. Defaults to False.
    """
    return Order.objects.get(owner=user, pk=order_id).to_dict(include_line_items)

@ag.tool
def list_orders(
    user,
    status: Literal["pending", "fulfilled", "cancelled"] = "pending",
    limit: int = 20,
) -> list[dict]:
    """List the user's orders, newest first.

    Args:
        status: Filter orders by fulfilment status.
        limit: Maximum number of orders to return.
    """
    return list(Order.objects.filter(owner=user, status=status)[:limit].values())

Accepted headers: Args:, Arguments:, Parameters:, Params: (case-insensitive). Multi-line descriptions (indented continuation lines) are joined with a space. Type annotations in name (type): form are ignored — the SDK already has the type from the signature.

#Option 2 — ag.Param explicit override

Use ag.Param when you need to override a single parameter independently without rewriting the docstring, or when you want to attach examples that the docstring format can't express. ag.Param always wins over the docstring Args: entry for the same parameter — other parameters in the same function still fall through to the docstring.

from typing import Annotated, Literal
import agentgraft as ag

@ag.tool
def list_orders(
    user,
    status: Annotated[
        Literal["pending", "fulfilled", "cancelled"],
        ag.Param("Filter orders by fulfilment status"),
    ] = "pending",
    limit: Annotated[int, ag.Param("Maximum number of orders to return", examples=[10, 50])] = 20,
) -> list[dict]:
    """List the user's orders, newest first."""
    return list(Order.objects.filter(owner=user, status=status)[:limit].values())

The examples list (optional) is merged into the parameter's JSON schema entry. Literal types automatically generate an enum constraint — combine them with ag.Param to tell the model both what values are valid and what they mean.

#@ag.agent

Declares an agent: a named persona with instructions, a list of tools, and optional handoff targets. The function body isn't executed; only its docstring matters (it becomes the agent's system prompt fragment) and the decorator arguments. Naming a function def helper(): and decorating it produces an agent named helper.

@ag.agent(entry=True, tools=[list_datasets, get_dataset])
def router():
    """You route incoming messages. For greetings or small talk,
    answer directly in one line. For data questions, hand off to
    ``data_helper`` immediately."""


@ag.agent(tools=[list_datasets, get_dataset, recent_queries])
def data_helper():
    """You help users explore their datasets. Always call a tool
    before answering; never invent values."""


@ag.agent(handoffs=["router"])
def fallback():
    """You handle anything off-topic. Politely redirect back to
    what the app actually does, then hand off to router."""

Exactly one agent per graft must be entry=True. Handoffs are referenced by name (string) so you can declare them in any order — the registry resolves them at sync time and rejects unknown names with a clear error.

Use sub_agents=[...] when you want one agent to delegate a task to another and receive a result back, rather than transferring the conversation. The sub-agent runs its own tool loop internally and returns a string; the calling agent stays in control and can use that result in its next step. This is the orchestrator pattern — contrast it with handoffs=, which hands the conversation over entirely.

@ag.agent(sub_agents=["researcher"], entry=True)
def advisor():
    """You advise users on their portfolio. When you need current market
    data or detailed research on a symbol, delegate to 'researcher' and
    use its result in your answer. Compose the final response yourself.
    """


@ag.agent(tools=[search_news, get_price_history])
def researcher():
    """You gather factual data. Call the tools you have, then return
    a concise factual summary for the advisor to use.
    """

An optional model argument pins the agent to a specific Cailos model codename. The default auto hands routing to Cailos; auto:speed and auto:cost are Cailos's speed- and cost-biased routes. Concrete ids like claude-opus-4-7, claude-sonnet-4-6, claude-haiku-4-5, or gpt-5 also work — whatever string you pass is sent to Cailos verbatim, so the source of truth for valid codenames is Cailos's model list. Invalid codenames surface as Cailos errors at request time; AgentGraft itself does no validation.

#Private chat

Mark one agent private=True and the widget gets an incognito chat: a ghost button in the header opens a conversation with that agent that AgentGraft does not save. No conversation, no messages, no long-term memory, no titles or recaps. The widget holds the transcript in the page and sends the recent turns with each message, so leaving the chat or reloading the page discards it.

@ag.agent(private=True, tools=[search_docs])
def incognito():
    """You answer questions the visitor would rather not keep.
    Use search_docs for anything about the product."""

• At most one private agent per graft, and it cannot be the entry point.
• It cannot declare handoffs or sub_agents, and no other agent can hand off to it or call it. Sync rejects all of these.
• It sees the graft prompt (identity, voice, scope, policies, references), the visitor's display name and the current date. It gets no memory tools and no stored facts, even when memory_enabled is on. Its own tools work as usual.
• Each message carries the most recent turns that fit in half the graft's context_window_tokens, up to 40 messages. Older turns drop out of the agent's view.
• Its spend counts toward visitor_quota_usd like any other turn.

#ag.graft(...)

Sets the graft-level prompt context. Every agent's system prompt is composed at runtime by stacking these blocks before the agent's own instructions, so the same identity, voice, scope, and policies apply across every conversation.

ag.graft(
    name="Acme",
    identity="You are Acme's in-app assistant.",
    product_context=(
        "Acme is a B2B analytics platform. Users own datasets,"
        " run queries, and share dashboards with their teams."
    ),
    voice="Be concise. Use British English. Cite the dataset name when quoting figures.",
    scope=(
        "IN SCOPE: dataset queries, dashboard navigation, query syntax help.\n"
        "OUT OF SCOPE: financial advice, anything off-platform."
    ),
    policies=[
        ag.policy("agentgraft/no-hallucination"),
        ag.policy("agentgraft/cite-sources"),
        "Never reveal API keys or any value beginning with sk_.",
    ],
    greeting="Ask me about your data.",
    example_prompts=["What datasets do I have?", "Show me yesterday's queries"],
)

All fields are optional. The dashboard's read-only graft page renders each one as a labelled section so you can see exactly what the runtime is composing for every conversation.

#Runtime features & limits on ag.graft

Alongside the prompt fields, ag.graft(...) accepts runtime-feature and limit kwargs that tune memory, recap, context windows, and spend ceilings per graft. All of them are tri-state — leave them as None (the default) to preserve whatever the server currently has, pass a concrete value to overwrite on the next sync. This mirrors how the name field already works: absent means "don't touch", present means "set explicitly".

• memory_enabled: bool — when True, the Dreamer curates a persistent fact pool for each visitor after every compaction, and the recall agent reads those facts on every subsequent turn. Default: False on fresh grafts. Turn on for any graft with returning visitors who benefit from cross-session memory.
• recap_enabled: bool — when True, a short "where you left off" summary is pre-generated after the conversation has been idle. Shown as a banner when the visitor reopens the thread, dismissed the moment they send their next message. Default: False.
• recap_min_gap_minutes: int — idle-time threshold before the recap sweep considers a conversation ready. Minimum 30 (the sweep floor); shorter values are rejected at sync time. Default: 30.
• recap_message_pairs: int — how many user + assistant pairs the recap summariser reads. Default: 10.
• recap_min_message_pairs: int — minimum pairs that must exist for a recap to be worth generating. Conversations shorter than this are skipped. Default: 3.
• context_window_tokens: int — overrides the context window ceiling used for conversation token profiling and accounting in the widget. Default: 32,000 tokens (valid range: 4,096 to 512,000).
• max_message_chars: int — hard character limit on every incoming user message. Rejects oversized input at the API boundary. Default: 10,000 chars (valid range: 1 to 10,000).
• visitor_quota_usd: Decimal | float — rolling spend ceiling in USD per visitor across all conversations, including private chats and replies the visitor stopped early. Default: $5.00.
• visitor_quota_window_days: int — duration of the rolling spend window in days. Older usage rolls off continuously. Default: 7 days (valid range: 1 to 365).
• image_uploads_enabled: bool — when True, visitors can drop, paste or pick up to 5 images per message (a GIF is sent as a few still frames and counts as one). The widget offers it only when the agent answering next can see images; agents whose model can't get a text description of each image instead. Images are encrypted at rest under the visitor's key, and each description is charged to the visitor's quota. On Safari the host page's CSP needs 'wasm-unsafe-eval' (see Embedding the widget). Default: False.
• voice_enabled: bool — when True, the widget offers voice: replies read aloud, dictation, and hands-free listening for visitors who opt in. Reply text and visitor audio go to the speech providers (Deepgram, or the visitor's browser), and read-aloud is charged to the visitor's quota. When False the widget shows no voice controls and the speech endpoints refuse. Default: False.
• followups_enabled: bool — when True, each reply ends with two or three suggested next questions the visitor can click. Adds one cheap model call per turn, made before the reply finishes. Default: False.
• palette: str — the widget's accent colour, so it matches your app: buttons, links, and the face's eyes and juice. One of indigo, orchid, rose, coral, amber or jade (ag.PALETTES); each has a dark and a light variant that follows the widget's theme. An unknown name fails at ag.graft(...). Default: "indigo".
• visitor_palette_enabled: bool — when True, visitors can pick their own palette under Personalisation in the widget's settings, over yours. Their pick stays in their browser, and picking yours again puts them back on it. Default: False.

from decimal import Decimal

ag.graft(
    name="Acme",
    identity="You are Acme's in-app assistant.",
    # Runtime features & limits — declare once, stays until you change it.
    context_window_tokens=32_000,
    max_message_chars=10_000,
    visitor_quota_usd=Decimal("5.00"),
    visitor_quota_window_days=7,
    memory_enabled=True,
    recap_enabled=True,
    recap_min_gap_minutes=60,        # only recap after an hour of silence
    recap_message_pairs=15,
    image_uploads_enabled=True,      # visitors can attach images
    voice_enabled=True,              # read aloud + talk instead of type
    followups_enabled=True,          # suggest questions after each reply
    palette="jade",                  # the widget's accent colour
    visitor_palette_enabled=True,    # visitors may pick their own
)

#ag.policy(codename)

References a reusable system-prompt fragment from the AgentGraft Policy Shop. The sync command fetches the policy body by codename at push time and substitutes it into the composed policies text — the runtime never sees the codename, only the resolved body.

Browse the catalogue at agentgraft.com/dashboard/policies/. Common codenames: agentgraft/no-hallucination, agentgraft/cite-sources, agentgraft/concise, agentgraft/secrets-redaction, agentgraft/mermaid-strict.

#Clickable references: ag.reference and ag.ref

Let the assistant link to objects in your app — documents, projects, people — as clickable chips inside its replies. Declare each type once with ag.reference(...), then have your tools return ag.ref(type, id, label) for every object they hand back. You can declare as many types as you need.

ag.reference(
    "block",
    label="Block",
    description="A document in the user's knowledge base.",
    url="/blocks/{id}",   # fallback when your JS doesn't handle the click
    icon="document",      # document | folder | user | calendar | link
)

@ag.tool
def search_blocks(user, query: str) -> list[dict]:
    """Search the user's blocks."""
    return [
        {"ref": ag.ref("block", b.id, b.title), "snippet": b.snippet}
        for b in Block.objects.search(user, query)
    ]

ag.ref returns text like [Q3 Plan](ref:block/42). The assistant is told to paste refs verbatim, and the widget renders each one as a chip. Clicking a chip dispatches a cancelable agentgraft:reference event on window. Call preventDefault() to open the object in-app; otherwise the widget navigates to url with {id} filled in. In fullscreen or on mobile, the widget collapses after a click so the object isn't hidden behind it.

window.addEventListener("agentgraft:reference", (event) => {
  const { type, id, label } = event.detail;
  if (type !== "block") return;
  event.preventDefault();   // skip the url fallback
  openBlock(id);            // your app's own navigation
});

#Multiple profiles

One host app can run several chat experiences side by side. Add another entry to AGENTGRAFT['profiles'], decorate the tools and agents for it with profile="<name>", and the registries stay physically separate inside the SDK. A marketing tool can never be called against an authenticated user, and vice versa.

# settings.py
AGENTGRAFT = {
    "config_module": "myapp.agentgraft_config",
    "profiles": {
        "default": {  # the in-app helper, authenticated users
            "api_base":       "https://agentgraft.com",
            "api_key":        os.environ["AGENTGRAFT_API_KEY"],
            "public_key":     os.environ["AGENTGRAFT_PUBLIC_KEY"],
            "graft_id":       os.environ["AGENTGRAFT_GRAFT_ID"],
            "webhook_secret": os.environ["AGENTGRAFT_WEBHOOK_SECRET"],
        },
        "marketing": {  # the public landing-page bot, anonymous visitors
            "api_base":       "https://agentgraft.com",
            "api_key":        os.environ["AGENTGRAFT_MARKETING_API_KEY"],
            "public_key":     os.environ["AGENTGRAFT_MARKETING_PUBLIC_KEY"],
            "graft_id":       os.environ["AGENTGRAFT_MARKETING_GRAFT_ID"],
            "webhook_secret": os.environ["AGENTGRAFT_MARKETING_WEBHOOK_SECRET"],
        },
    },
}

# myapp/agentgraft_config.py
@ag.tool                            # default profile, authenticated
def list_my_datasets(user: User): ...

@ag.tool(profile="marketing")       # marketing profile, anonymous
def what_is_acme(user: dict): ...

@ag.agent(entry=True, tools=[list_my_datasets])
def helper(): ...

@ag.agent(profile="marketing", entry=True, tools=[what_is_acme])
def landing_bot(): ...

#manage.py agentgraft_sync

The sync command serialises every registry the config module populated and POSTs the bundle to the matching graft's /api/v1/admin/grafts/<id>/sync/ endpoint. The push is atomic on the server side — either every tool, agent, and prompt change lands together, or none of them do.

# Sync every declared profile (default behaviour)
python manage.py agentgraft_sync

# Restrict to one profile
python manage.py agentgraft_sync --profile marketing

# Print the payload, don't send it (great for debugging)
python manage.py agentgraft_sync --dry-run

# CI-friendly: don't crash on missing credentials
python manage.py agentgraft_sync --skip-if-unconfigured

# Override retry count on flaky networks (default: 3)
python manage.py agentgraft_sync --retries 5

The --skip-if-unconfigured flag is the one most deployment pipelines need — see the Deployment page for the full Railway preDeploy story. The TL;DR is that it lets your first deploy succeed before you've pasted credentials into env vars, and silently no-ops profiles that aren't fully configured yet instead of failing the build.

#The webhook view

The Django adapter ships a single view that handles every incoming tool call. Mount its URL include and the SDK registers /agentgraft/webhook/ and /agentgraft/webhook/<profile>/ automatically.

# urls.py
from django.urls import include, path

urlpatterns = [
    path("agentgraft/", include("agentgraft.django.urls")),
]

The view verifies the HMAC signature on every incoming request, rejects requests whose signed timestamp is more than 5 minutes from the host clock in either direction (replay protection), looks up the matching tool by name in the profile's registry, checks the arguments fit it, runs it with the resolved user through the profile's tool-call hooks, and returns the JSON-serialised result. You don't write any view code — the SDK handles the entire request lifecycle.

#Tool-call hooks

@ag.on_tool_call registers a hook that wraps every tool call a profile's webhook runs: the place for what applies to all your tools at once, like an audit trail, metrics or a kill switch. The hook gets the call and proceed; calling proceed() runs the rest of the chain and the tool and returns the result, and whatever the hook returns is the call's result.

import time

import agentgraft as ag


@ag.on_tool_call
def audited(call: ag.ToolCall, proceed):
    started = time.monotonic()
    outcome = "error"
    try:
        result = proceed()
        outcome = "ok"
        return result
    finally:
        AuditLog.objects.create(
            user=call.user,
            agent=call.agent,
            tool=call.tool,
            request_id=call.request_id,
            outcome=outcome,
            duration_ms=round((time.monotonic() - started) * 1000),
        )

• call is an ag.ToolCall: tool, arguments (a read-only view), user (what your loader returned), user_external_id, is_anonymous, private, agent (the agent that made the call, by its name in your config), request_id and profile.
• A hook can act around the call, replace its result, answer without running the tool (return without calling proceed()), or refuse it: raise ag.WebhookError(403, "search is disabled") sends that status and message back, and the model sees the error (a 502, 503 or 504 makes AgentGraft retry the call once). Any other exception from a hook is a 500.
• Hooks run only for calls that reach a tool: verified, with the user loaded and arguments that fit. The first registered is outermost; each profile has its own (@ag.on_tool_call(profile="public")).
• ag.current_tool_call() returns the call in progress to hooks, the tool and anything they call, and None outside a webhook call.

#The template tag

One tag, no arguments, in any Django template:

{% load agentgraft %}
{% agentgraft_widget %}

By default it renders the default profile. Pass profile="marketing" to render a different profile on a different page. The tag emits a single <script type="module"> element with the public key, the user external id, and an HMAC signed by the matching profile's webhook secret. The widget bundle resolves these from the script's data attributes at boot.

#Programmatic Chat & Omnichat (Client)

Beyond the drop-in browser widget, AgentGraft provides a first-class programmatic client (agentgraft.Client) with zero external dependencies. It supports two primary integration models:

• Autonomous Omnichat (zero thread management needed)
• Custom UI & Thread Lifecycle (for custom frontends & mobile apps)

#1. Autonomous Omnichat (Slack, WhatsApp, API Workers)

For most backend integrations (Slack bots, WhatsApp handlers, SMS bots, Celery tasks, or backend API proxies), you do not need to manage or store conversation IDs. Passing auto_route=True lets AgentGraft handle the entire lifecycle autonomously:

from agentgraft import Client

client = Client.from_profile("default")

# One line to chat — AgentGraft matches existing topic threads,
# pre-fetches long-term memories, dispatches tools, and auto-titles the thread.
response = client.chat(
    user_id="slack:U123456",
    message="What are my top 3 active projects?",
    auto_route=True,
    visitor_name="Alice",
)

print(f"Assigned Thread: {response['conversation_id']}")
print(f"Assistant: {response['text']}")

#2. Explicit Thread Continuity & Streaming

When you already have a conversation in flight or want real-time token streaming, pass the conversation_id explicitly:

# Real-time SSE streaming inside a specific thread
for event in client.stream_chat(
    user_id="slack:U123456",
    conversation_id=response["conversation_id"],
    message="Can you detail project #2?",
):
    if event["type"] == "token":
        print(event["text"], end="", flush=True)
    elif event["type"] == "tool_call":
        print(f"\n[Calling tool: {event['name']}]")
    elif event["type"] == "done":
        print("\n[Finished]")

#3. Custom UI & Thread Lifecycle Management

If you are building your own custom user interface (e.g. an iOS/Android mobile app, custom React/Vue chat widget, or admin dashboard), Client gives you full programmatic control over conversations, messages, and feedback:

# List all threads for a user
threads = client.list_conversations(user_id="user:42")

# Fetch full message history of a thread
messages = client.get_messages(user_id="user:42", conversation_id="convo-uuid")

# Semantic + substring search across visitor history
results = client.search_messages(user_id="user:42", query="quarterly invoice")

# Submit RLHF thumbs up/down rating on an assistant message
client.submit_feedback(user_id="user:42", message_id="msg-uuid", positive=True)

# Programmatic long-term memory CRUD
memories = client.list_memories(user_id="user:42")
client.create_memory(user_id="user:42", content="Prefers monthly billing")
client.update_memory(user_id="user:42", memory_id="fact-uuid", content="Prefers annual billing")
client.delete_memory(user_id="user:42", memory_id="fact-uuid")

#Testing

Tools are plain Python functions — call them directly in tests with a stub user object. The agentgraft.testing module gives you a pytest fixture and two utilities for testing the registry and schema in addition to raw function logic.

# conftest.py — make ag_reset available project-wide
pytest_plugins = ["agentgraft.testing"]

from agentgraft.testing import call_tool, tool_schema

def test_get_order_returns_correct_fields(db, ag_reset):
    user = make_test_user()
    result = call_tool(get_order, user, order_id="abc123")
    assert result["id"] == "abc123"

def test_get_order_schema_has_required_order_id():
    schema = tool_schema(get_order)
    assert "order_id" in schema["required"]
    assert schema["properties"]["order_id"]["description"] != ""

def test_list_orders_plain_call(db):
    # Tools are plain functions — no framework required
    user = make_test_user()
    orders = list_orders(user, status="pending")
    assert all(o["status"] == "pending" for o in orders)

call_tool(fn_or_name, user, **kwargs) — call a registered tool by function reference or string name, with user injected as the first argument, exactly as the webhook handler would. Accepts the same **kwargs the LLM would supply. The tool runs alone, without the profile's tool-call hooks; test a hook by calling it with an ag.ToolCall and a proceed of your own.

tool_schema(fn_or_name) — return the JSON schema dict the SDK would push at sync time. Use it to assert that parameter descriptions, types, and required fields are exactly right before they reach production.

ag_reset fixture — resets the registry before and after each test, preventing decorator registrations from leaking between test cases.