agentgraft 0.1.0

Python SDK for AgentGraft — define your chat config in code, sync to the hosted service.

agentgraft

Python SDK for AgentGraft. Define your chat configuration in code, sync it to the hosted service, and embed a chat widget on your site with one template tag.

pip install agentgraft           # core SDK (framework-agnostic)
pip install agentgraft[django]   # + Django management command, webhook view, template tag

The PyPI distribution and the import name are both agentgraft. Django glue lives at agentgraft.django; everything else is framework-agnostic.

---

How it fits together

Three actors:

• End user — talks to the widget in their browser.
• Host app — your Django project. Owns user identity, owns the data, owns tool execution.
• AgentGraft — hosted service. Owns the chat UI, the LLM loop, and the conversation storage.

The SDK is the bridge. You write one Python file declaring your tools, agents, and prompt context. You run one command to push that file to AgentGraft. You drop one template tag wherever the chat should appear. When the LLM wants to call a tool, AgentGraft signs an HTTP request to your host app's webhook URL; the SDK receives it, verifies the signature, runs your function, returns the result.

The host app's database is never read by AgentGraft directly. AgentGraft only ever sees tool schemas — never tool implementations.

---

Quickstart (Django)

1. Install

pip install agentgraft[django]

2. Add to INSTALLED_APPS

# settings.py
INSTALLED_APPS = [
    # ...
    "agentgraft.django",
]

3. Configure

# settings.py
AGENTGRAFT = {
    "config_module": "myapp.agentgraft_config",
    "profiles": {
        "default": {
            "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"],
        },
    },
}

The four credentials come from creating a graft on AgentGraft. The api_key is shown only once at creation — store it as a secret.

4. Mount the webhook URL

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

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

This exposes POST /agentgraft/webhook/ for tool dispatches. Configure that URL as your graft's webhook_url (either at graft creation or inside ag.graft(webhook_url=...)).

5. Write the config

Create myapp/agentgraft_config.py:

import agentgraft as ag
from myapp.models import User


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


@ag.tool
def list_datasets(user: User) -> list[dict]:
    """List the user's datasets."""
    return list(user.datasets.values("id", "name"))


@ag.tool
def get_dataset(user: User, dataset_id: int) -> dict:
    """Fetch one dataset by id."""
    ds = user.datasets.filter(pk=dataset_id).first()
    return {"id": ds.id, "name": ds.name} if ds else {}


@ag.agent(entry=True, tools=[list_datasets, get_dataset])
def helper():
    """You help users explore their datasets. Be concise."""


ag.graft(
    name="MyApp",
    identity="You are MyApp's assistant.",
    voice="Be direct. Skip filler.",
    policies=[
        ag.policy("agentgraft/no-hallucination"),
        ag.policy("agentgraft/concise"),
    ],
    greeting="Ask me about your datasets.",
    example_prompts=[
        "How many datasets do I have?",
        "Show me the most recent one",
    ],
)

6. Sync

python manage.py agentgraft_sync --dry-run   # inspect the payload
python manage.py agentgraft_sync             # push it

7. Embed the widget

In any Django template:

{% load agentgraft %}
{% agentgraft_widget %}

That's the whole integration.

---

Settings reference

Every key lives under the AGENTGRAFT dict in settings.py.

AGENTGRAFT = {
    "config_module": "myapp.agentgraft_config",  # optional, default: "agentgraft_config"
    "profiles": {
        "default": {
            "api_base":       "https://agentgraft.com",
            "api_key":        "sk_live_...",
            "public_key":     "pk_live_...",
            "graft_id":       "uuid",
            "webhook_secret": "whsec_...",
            "webhook_url":    "https://myapp.com/agentgraft/webhook/",  # optional
        },
    },
}

Key	Required by	Description
config_module	AppConfig.ready	Dotted path to your config module. Defaults to "agentgraft_config" (a top-level agentgraft_config.py).
profiles.<name>.api_base	sync, widget	Origin of the AgentGraft service.
profiles.<name>.api_key	sync	Bearer token for /api/v1/admin/grafts/{id}/sync/. Shown once at graft creation.
profiles.<name>.public_key	widget	Browser-safe key embedded in the widget script tag.
profiles.<name>.graft_id	sync, widget	UUID of the graft.
profiles.<name>.webhook_secret	webhook view, widget HMAC	HMAC signing secret used by both AgentGraft (signing tool dispatches) and the host (signing widget identity).
profiles.<name>.webhook_url	sync (optional)	Where AgentGraft posts tool calls. May also be set via ag.graft(webhook_url=...). Profile config wins if both are set.

Pull api_key and webhook_secret from environment variables, never from a checked-in settings file.

---

Decorator reference

import agentgraft as ag

@ag.user_loader

Resolves an opaque external id back to your user model. Called once per webhook request before any tool runs.

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

The widget passes external ids prefixed with user: for authenticated visitors and session: for anonymous visitors. Strip the prefix yourself or branch on it. Optionally accept an is_anonymous: bool keyword:

@ag.user_loader
def load_user(external_id: str, *, is_anonymous: bool = False):
    if is_anonymous:
        return {"session_key": external_id.removeprefix("session:")}
    return User.objects.get(pk=external_id.removeprefix("user:"))

Raise any exception (e.g. DoesNotExist) to fail the lookup — the webhook view turns it into a 403. Only one @ag.user_loader per profile.

@ag.tool

Defines a function the LLM can call. The function runs inside your host app, not on AgentGraft's servers.

@ag.tool
def search_assets(user: User, query: str, limit: int = 5) -> list[dict]:
    """Search the catalogue for assets matching the query."""
    return list(user.assets.filter(name__icontains=query)[:limit].values())

Rules:

• The first parameter is always user and is never sent to the LLM.
• Every other parameter must have a Python type hint. The SDK builds a JSON schema from those hints — no hint, no decoration.
• The docstring becomes the description the LLM sees, or pass description="...".
• A parameter with a default is optional in the schema; without one, it's required.
• Tool names must be unique within a profile.

Parameter descriptions

The LLM fills in tool arguments using the parameter name, its type, and — if provided — a natural-language description. Descriptions are one of the highest-leverage ways to improve tool call accuracy in a multi-agent system.

There are two ways to add them. Both produce the same result in the JSON schema the LLM receives. You can mix them freely in the same function.

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

Write a standard Args: block in the docstring. The SDK extracts it automatically at sync time — no extra imports required:

@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.
    """
    ...

The Args: section is stripped from the tool-level description before it reaches the LLM — the model sees a clean summary in the tool description and targeted guidance in each parameter's schema field.

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 knows the type from the signature.

Option 2 — ag.Param explicit override

Use ag.Param when you need to override a single parameter without rewriting the full docstring, or when you want to attach examples:

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."""
    ...

Priority: 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.

---

Good descriptions answer: what is this value, where does it come from, and when should I omit it. They don't repeat type information already visible in the schema.

Without description	With description
dataset_id: int	"Numeric ID returned by list_datasets"
query: str	"Free-text search across ticker and company name"
since: str	"ISO 8601 date string — only return events after this date"

@ag.agent

Defines an agent — a system prompt with an allowed tool set and optional handoff targets.

@ag.agent(
    entry=True,
    tools=[search_assets, get_asset],
    handoffs=["analyst"],
    model="auto",
)
def concierge():
    """You are MyApp's concierge. Route the user to the right agent."""

The function body is never executed. Only the docstring matters — it becomes the agent's instructions. Kwargs:

Param	Type	Default	Meaning
entry	bool	False	Mark as the conversation's entry point. At most one per profile.
tools	list	[]	Tool function references or string names.
handoffs	list	[]	Names of agents this one can hand off (transfer) to.
sub_agents	list	[]	Names of agents to invoke as tool calls (orchestrator pattern).
model	str	"auto"	Any Cailos model codename — e.g. auto, auto:speed, auto:cost, claude-opus-4-7, gpt-5. Passed to Cailos verbatim; invalid values surface as Cailos errors.

handoffs vs sub_agents — use handoffs when you want to transfer the conversation to another agent (the caller steps back). Use sub_agents when you want to delegate a task and get a result back — the calling agent stays in control and can use the result in its next step. Sub-agents run their own tool loop internally and return a string.

ag.graft(...)

Sets the graft-level prompt context and the widget welcome screen.

ag.graft(
    name="MyApp",
    identity="You work inside MyApp's customer dashboard.",
    product_context="MyApp is a portfolio analytics tool.",
    voice="Be concise. Reach for examples over abstractions.",
    scope="Only answer questions about the user's own portfolio.",
    policies=[
        ag.policy("agentgraft/no-hallucination"),
        ag.policy("agentgraft/cite-sources"),
        "Never give specific buy/sell advice.",
    ],
    webhook_url="https://myapp.com/agentgraft/webhook/",
    greeting="Ask me about your portfolio.",
    example_prompts=[
        "How is my portfolio performing this week?",
        "Which positions are down the most?",
    ],
)

The five prompt fields (identity, product_context, voice, scope, policies) get composed into every agent's system prompt at runtime. The two widget fields (greeting, example_prompts) drive the empty state visitors see before sending their first message — both fall back to built-in defaults if blank.

policies accepts either a plain string OR a list mixing free-form strings with ag.policy(...) references. References are resolved to their full body text from the AgentGraft Policy Shop at sync time.

Runtime features are also configured via ag.graft() and applied at sync time:

Field	Type	Default	Description
memory_enabled	bool	False	Enable the long-term memory system. When on, the Dreamer curates a persistent fact pool for each visitor after every compaction; memory_finder and memory_manager become available as built-in tools the LLM can call.
recap_enabled	bool	False	Pre-generate a short "where you left off" recap after the conversation has been idle. Shown as a banner when the visitor reopens the thread; dismissed the moment they send their next message.
recap_min_gap_minutes	int	30	Minutes of inactivity required before a recap is generated. Minimum: 30 (the sweep floor).
recap_message_pairs	int	10	How many user + assistant message pairs to include in the recap window.
recap_min_message_pairs	int	3	Minimum pairs that must exist before a recap is worth generating. Conversations shorter than this are skipped.

Example:

ag.graft(
    name="MyApp",
    identity="You are MyApp's assistant.",
    greeting="Ask me about your datasets.",
    memory_enabled=True,
    recap_enabled=True,
    recap_min_gap_minutes=60,  # only recap after 1 h of silence
)

ag.policy(codename)

Reference a curated policy from the AgentGraft Policy Shop. Returns an opaque sentinel that the sync command resolves into the actual prompt text by fetching from the public catalogue.

ag.policy("agentgraft/mermaid-strict")
ag.policy("agentgraft/no-hallucination")
ag.policy("agentgraft/secrets-redaction")

Browse the catalogue at https://agentgraft.com/dashboard/policies/ or fetch the list yourself: GET /api/v1/policies/.

---

Multiple profiles

A single host app can declare multiple chat experiences in one config file by passing profile= to the decorators:

@ag.tool(profile="public")
def what_is_myapp(user) -> str:
    """Return a one-paragraph pitch."""
    return "MyApp is..."

@ag.agent(profile="public", entry=True, tools=[what_is_myapp])
def public_concierge():
    """You greet visitors on the marketing site."""

ag.graft(profile="public", name="MyApp Public", ...)

Each profile is an independent registry — a public-profile tool can never be called against an authenticated-profile user, and vice versa. Add the profile to settings:

AGENTGRAFT = {
    "profiles": {
        "default": {...},
        "public": {...},
    },
}

Sync individual profiles or all at once:

python manage.py agentgraft_sync                   # every declared profile (default)
python manage.py agentgraft_sync --profile public  # one profile only

In templates, name the profile to render:

{% agentgraft_widget %}                    {# default #}
{% agentgraft_widget profile="public" %}   {# named #}

---

Anonymous visitors

Anonymous visitors get persistent conversation history via Django sessions. The template tag handles this automatically:

• Authenticated user → external id is user:{user.pk}, is_anonymous=False.
• Anonymous visitor → external id is session:{session.session_key}, is_anonymous=True. The tag forces the session to exist if it doesn't yet, so the same browser stays on the same conversation across reloads.

Both paths sign the external id with HMAC-SHA256 server-side using the profile's webhook_secret. The widget sends the signature back as X-AgentGraft-User-HMAC on every API call; AgentGraft verifies it and rejects forged identities.

Your @ag.user_loader receives both the external id and the is_anonymous keyword — branch on it to return either a real User row or a guest sentinel.

---

manage.py agentgraft_sync

python manage.py agentgraft_sync               # push every declared profile
python manage.py agentgraft_sync --profile X   # push profile X only
python manage.py agentgraft_sync --dry-run     # print the payload, don't send
python manage.py agentgraft_sync --retries 5   # retry up to 5 times on transient errors

What it does:

1. Reads the named profile's settings. Refuses to run if api_base, api_key, or graft_id are missing.
2. Reads the registry populated at app startup. Refuses to run if it's empty (usually means the config module failed to import).
3. Resolves any ag.policy(...) references by fetching from the public Policy Shop endpoint.
4. Composes the resolved bodies + free-form strings into a single policies text block.
5. Builds the sync payload (tools, agents, graft metadata).
6. POSTs to {api_base}/api/v1/admin/grafts/{graft_id}/sync/ with Authorization: Bearer {api_key}.
7. Prints a summary of added / updated / removed tools and agents.

Sync is declarative: tools and agents that exist on the server but not in your local config are deleted. Push the entire registry, every time. The server applies it atomically — either every diff lands or none do.

---

The webhook view

Mounted by path("agentgraft/", include("agentgraft.django.urls")). Exposes:

• POST /agentgraft/webhook/ — default profile
• POST /agentgraft/webhook/<profile>/ — named profile

The view:

1. Verifies the X-AgentGraft-Signature HMAC against the profile's webhook_secret. Bad signature → 403.
2. Parses the JSON body. Bad JSON → 400.
3. Looks up the tool by name. Unknown tool → 404.
4. Calls your user_loader. Any exception → 403.
5. Calls the tool function with the user + LLM-supplied arguments. Argument mismatch → 400. Other exceptions → 500.
6. Returns {"result": <whatever the tool returned>} as JSON.

The view is csrf_exempt and require_POST. You write none of this — just register your tools and a user_loader.

---

The wire contract

For debugging what AgentGraft sends to your webhook.

Request body:

{
  "tool": "list_datasets",
  "arguments": { "limit": 5 },
  "user_external_id": "user:42",
  "is_anonymous": false
}

Headers:

Content-Type: application/json
X-AgentGraft-Signature: <hex sha256 hmac of the raw body, using webhook_secret>

Success response (200):

{ "result": <whatever the tool returned> }

Error responses: {"error": "<message>"} with one of:

Code	Meaning
400	Malformed JSON, missing tool, bad arguments
403	Invalid HMAC, missing signature, or user_loader raised
404	Tool name not in the registry
500	Tool function raised, or webhook secret not configured

---

Migration import API

Migrating from another chat platform, or rolling your own history into a fresh graft? Backfill end users + conversations + messages with a single POST so the Dreamer can extract long-term memory and the recall agent can retrieve it. Without this, every migrated visitor starts "cold" — empty fact pool, no history — and their experience regresses.

Endpoint — POST /api/v1/admin/grafts/{graft_id}/import/

Auth — your existing graft API key, same Authorization: Bearer sk_live_… header used by agentgraft_sync.

Request body

{
  "dry_run": false,
  "users": [
    {
      "external_id": "your-stable-user-id",
      "display_name": "Optional display name",
      "conversations": [
        {
          "title": "Optional",
          "messages": [
            {"role": "user",      "content": "hi",    "created_at": "2026-01-15T10:00:00Z"},
            {"role": "assistant", "content": "hello", "created_at": "2026-01-15T10:00:05Z"}
          ]
        }
      ]
    }
  ]
}

Contract

• Purely additive. If an EndUser with this external_id already exists in the graft, the entire user entry is skipped with a warning — we never update existing rows.
• Your external_id is your problem. We store it verbatim. Stability across migrations is on you; if it changes, you'll fork the visitor's identity.
• Original timestamps are preserved on every message, so the Dreamer reads the history in real chronological order.
• Token counts are recomputed from message content for consistency; cost_usd is always 0 for imported messages (we didn't serve them, we don't bill for them).
• dry_run: true validates + reports counts without writing anything and without enqueueing any background jobs.

Role mapping (silent)

Source systems use all kinds of role names. These are silently mapped to the canonical AgentGraft roles at import time:

Incoming role	Stored as
user, human, customer, end_user, enduser	user
assistant, bot, ai, gpt, model	assistant
tool, function	tool

Anything else is rejected as a validation error. system is explicitly not mapped — it's an internal orchestrator role and would let a migration payload inject instructions into a visitor's memory.

Alternation enforcement

Imported history must follow user → assistant alternation (tool messages between them are fine). If the source system lost an assistant reply and has two consecutive user messages — or a user message with no response at all — the stranded user message is dropped with a warning instead of creating a broken conversation.

If a conversation becomes empty after this cleaning, it's skipped with a warning. Empty conversations are never created.

Response

{
  "dry_run": false,
  "summary": {
    "users_created": 12,
    "users_merged": 1,
    "users_failed": 0,
    "conversations_created": 40,
    "conversations_skipped_empty": 3,
    "messages_created": 812,
    "messages_dropped_orphan_user": 5
  },
  "warnings": [
    {"user_external_id": "u-7", "conversation_index": 0, "reason": "conversation empty after cleaning, skipped"},
    {"user_external_id": "u-7", "conversation_index": 1, "message_index": 4,
     "reason": "orphan user message dropped (no assistant response)"}
  ],
  "errors": [
    {"user_external_id": "u-99", "reason": "unknown role 'overlord' at conversations[0].messages[2]"}
  ]
}

users_merged counts end-users whose (graft, external_id) row already existed: the row is reused and the supplied conversations are appended under it. Conversations themselves are never deduped, so avoid re-sending conversations you've already imported.

Per-user errors don't abort the whole batch — each user's work is wrapped in a savepoint, so one bad row only rolls back that user. Siblings succeed.

What happens after import

For grafts with memory_enabled=True, the Dreamer is automatically scheduled for every newly-created conversation as soon as the import commits. Embeddings are scheduled for every imported user and assistant message regardless of the memory toggle (embeddings power recall; they're independent of the facts pipeline). Both run asynchronously via Celery — the API response returns immediately and memory builds up in the background.

Size limits

There are no rate or size limits in the current release. Keep your batches reasonable (hundreds of users, thousands of messages) and chunk bigger imports client-side if needed.

---

Type-hint → JSON schema

Python annotation	JSON schema
str	{"type": "string"}
int	{"type": "integer"}
float	{"type": "number"}
bool	{"type": "boolean"}
list[T]	{"type": "array", "items": <T>}
dict	{"type": "object"}
Optional[T] / T | None	<T>, removed from required
Literal["a", "b"]	{"type": "string", "enum": ["a", "b"]}
Enum subclass	{"type": "string", "enum": ["val1", "val2", ...]}
Annotated[T, ag.Param("…")]	<T schema> + "description": "…"

A parameter with a default value is removed from required even if its annotation isn't Optional. Missing type hints are a hard error at decoration time — fail loud, not later in production.

---

Testing

Tools are plain Python functions. Call them directly with a stub user:

def test_list_datasets():
    user = make_test_user()
    result = list_datasets(user)
    assert all("id" in row for row in result)

For richer test coverage, use the agentgraft.testing module. It provides two utilities and a pytest fixture:

# conftest.py — enable the fixture project-wide
pytest_plugins = ["agentgraft.testing"]

from agentgraft.testing import call_tool, tool_schema

def test_get_order_calls_tool(db, ag_reset):
    user = make_test_user()
    # call_tool injects user as the first argument, just like the webhook does
    result = call_tool(get_order, user, order_id="abc123")
    assert result["id"] == "abc123"

def test_get_order_schema():
    schema = tool_schema(get_order)
    assert "order_id" in schema["required"]
    assert schema["properties"]["order_id"]["type"] == "string"

call_tool(fn_or_name, user, **kwargs) — call a registered tool by reference or by string name, injecting user as the first argument.

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

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

---

Without Django

The framework-agnostic core (registry, decorators, schema generation, webhook handler) is importable on its own. The Django adapter is a thin wrapper around agentgraft.handle_webhook, which you can call from any framework:

from agentgraft import handle_webhook, WebhookError

def my_flask_view():
    try:
        result = handle_webhook(
            body=request.get_data(),
            signature=request.headers.get("X-AgentGraft-Signature", ""),
            profile="default",
        )
    except WebhookError as exc:
        return {"error": exc.message}, exc.status_code
    return result, 200

The same pattern works for FastAPI, Starlette, Bottle, raw WSGI. Native adapters for other frameworks are not yet shipped.

---

Working from a source checkout

agentgraft.__version__ is sourced from installed package metadata via importlib.metadata. If you're running the SDK from a source checkout that hasn't been pip install-ed, the lookup falls through to a sentinel:

>>> import agentgraft
>>> agentgraft.__version__
'0.0.0+local'

This is by design — there's no dist-info directory to read from. Run pip install -e sdk/ (or your equivalent) to populate the metadata and pick up the real version string.

---

License

MIT. See LICENSE for the full text.