Running from a worker¶

Every worked example in these docs is HTTP-shaped, down to deps=AgentDeps(user=request.user). The larger surface for a lot of projects is the other one: a Celery task, a management command, a scheduled job — an agent run with nobody waiting on it.

There is no headless mode to switch on, and this page is not describing one. The same SpecToolset runs either way; a spec dispatched from a worker takes exactly the path it takes behind a view, because that path was never an HTTP one to begin with. What actually differs is two things: there is no request, and there is more load. Everything below follows from one of those.

The smallest run¶

Agent.run_sync works today and needs nothing from this package:

tasks.py

from celery import shared_task
from django.contrib.auth import get_user_model
from pydantic_ai import Agent

from django_spec_agents import AgentDeps
from django_spec_agents.spec_tools import SpecToolset

toolset = SpecToolset({"list_orders": list_orders_spec})
agent = Agent("anthropic:claude-opus-4-8", deps_type=AgentDeps, toolsets=[toolset])


@shared_task
def summarise_account(user_id: int) -> str:
    user = get_user_model().objects.get(pk=user_id)
    result = agent.run_sync(
        "Summarise this account's open orders.",
        deps=AgentDeps(user=user),
    )
    return result.output

Build the toolset and the agent once, at module scope. Both are reusable across runs — the toolset resolves its specs, validates its declarations and derives each tool's schema in the constructor, and repeating that per task is pure waste. Nothing per-request is closed over in either, which is what makes that safe.

The acting user is the whole authorization story¶

AgentDeps.user is not a convenience. It is the identity every permission_class, extend_queryset and scoping provider resolves against, and off HTTP it is the only thing that carries it.

Resolve it server-side, from something you trust: a task argument that is a primary key you put there, a row you already own. Never from the model's own output, and never from text the model was given — a run that can talk its way into a different user has no authorization boundary at all.

The run's request is not an identity

A request in the run's call_context is there so a callable or scoping provider reading request.META finds something plausible. Nothing downstream re-derives the acting user from it. Passing an authenticated request authorizes nothing, and leaning on it would put a second, invisible identity in the call.

Nobody is anonymous by accident¶

A view refuses an unauthenticated caller before the agent is reached. A worker has no such gate, so a task whose user_id is None runs with user=None and whatever the spec's permissions make of that. Decide it at the top of the task, where the answer is obvious, rather than in a permission class three layers down.

Concurrency: the one setting a worker genuinely needs¶

Every tool call goes through sync_to_async, which defaults thread_sensitive=True — and asgiref's single_thread_executor is a class attribute, so that is one thread for the whole process, shared by every toolset instance and every concurrent run in it.

Pydantic-AI runs function tools in parallel within a segment. Four 0.30s tool calls under one model step therefore take about 1.2s rather than 0.3s. In a chat turn calling one tool this is invisible. In a fan-out it is the ceiling.

from concurrent.futures import ThreadPoolExecutor

toolset = SpecToolset(
    SPECS,
    thread_sensitive=False,
    executor=ThreadPoolExecutor(max_workers=8),
)

Read this before flipping it

thread_sensitive=True is what keeps Django's thread-local database connections coherent. Setting it False moves dispatch onto a pool, and each of those threads opens and owns its own connection. That is fine — and it is what you want — provided you have accounted for it:

Your database can take max_workers more connections per worker process, or you front it with a pooler.
Nothing in your specs relies on being in the caller's transaction. An outer atomic() block in the task does not wrap work that ran on another thread.
You are not inside pytest.mark.django_db without transaction=True, where the test's wrapping transaction is invisible to other threads.

executor= is only consulted when thread_sensitive is False — that is asgiref's rule, not this package's. Setting it alone does nothing, silently.

Reading the logs afterwards¶

Nobody is watching a background run, so the log line is the whole record. Every tool call logs at DEBUG (WARNING for a permission denial, which is otherwise traceless — the run loop absorbs it into a message) on the django_spec_agents.spec_tools logger, and carries correlation fields in extra:

Field
`run_id`, `conversation_id`	which run this line belongs to
`run_step`, `tool_call_id`	where in that run
`run_input_tokens`, `run_output_tokens`, `run_requests`, `run_tool_calls`	the run's usage so far

Concurrent runs interleave, and without these the lines are indistinguishable. The usage fields are cumulative for the run rather than attributable to the call — a tool call spends no tokens itself — which is exactly what makes them useful here: they say where the budget stood when a long run went wrong. They ride on the timing line; the run_id / conversation_id / run_step / tool_call_id four are on every line the package emits, including the two WARNINGs a misbehaving call produces (a dispatch timeout, a result over its byte ceiling).

settings.py

LOGGING = {
    "version": 1,
    "loggers": {
        "django_spec_agents.spec_tools": {"level": "DEBUG", "handlers": ["json"]},
    },
}

Bounding the run is Agent.run's job, not the toolset's. Pass UsageLimits there. A toolset can only refuse the next tool call, which is the wrong instrument and a second place for the limit to live.

The limits are still readable from a tool call, though, and one thing is worth doing with them: ctx.usage_limits reaches an enforce_result_bytes override, so a project can taper how much a call is allowed to return as the run consumes its budget. That is shaping a result, not enforcing a limit — the run still stops where Agent.run says it does.

Bounding what comes back¶

A model reading a 40,000-row selector is a cost and latency problem before it is a correctness one, and in a fan-out it is multiplied by every concurrent run.

max_page_size= caps what one list call can return, however large a limit the model asks for. The clamp is reported back in the envelope, so the model is told it received less than it asked for rather than silently truncated.
max_result_bytes= (and tool_max_result_bytes= per tool) bounds the serialized payload, answering with a refusal the model can act on.
dispatch_timeout= bounds how long a call blocks the run. It does not stop the work — the thread runs to completion; the run stops waiting.

Changing the pipeline¶

Five overridable methods cover a tool call end to end. The front half — call_seeds and translate_exception — is documented under SpecToolset; the back half is shape_page, render_output and enforce_result_bytes. All five receive the live RunContext, which is the point: a run's typed deps are usually what needs to reach them.

The one worth knowing about before you need it is render_output. By default a result is rendered through the spec's presenter and projected for the agent audience. If a background job feeds one spec's output into the next, project the handles away and the next step has nothing to key on. Presenting with the kernel's present(spec, result) instead is the opt-out — and passing projection=None is not it, since that means "derive one from the spec".

Testing it¶

The run loop is where the interesting failures live, and it needs a model. django_spec_agents.testing ships the two doubles this package's own suite uses — no provider, no network, no key. The module imports pydantic-ai alone, so it works on an install without the [specs] extra too:

from pydantic_ai import Agent

from django_spec_agents import AgentDeps
from django_spec_agents.spec_tools import SpecToolset
from django_spec_agents.testing import tool_calling_model


async def test_the_task_calls_the_tool():
    toolset = SpecToolset({"list_orders": list_orders_spec})
    agent = Agent(tool_calling_model("list_orders"), deps_type=AgentDeps, toolsets=[toolset])

    result = await agent.run("go", deps=AgentDeps(user=user))

    assert result.output == "done"

Test the task as a whole this way rather than calling call_tool directly. Asserting on call_tool answers does the tool work; only a run answers whether the tools execute in-process, whether a ModelRetry is fed back for another attempt, and whether the toolset's instructions reached the model.

If your worker code is sync, drive it with agent.run_sync inside django_db(transaction=True) — the same shape production uses.

Progress, when there is no browser¶

AgentDeps.progress takes a ProgressReporter, and a spec that reports progress calls it wherever it runs. This package never constructs one: picking a sink would be picking a transport it does not own. Off HTTP, the useful sinks are the ones a worker already has — a task-state update, a row, a log line.