Agents & types

The unified Agent class plus the value types every runtime call accepts or returns.

from murmur import (
    Agent,
    AgentContext,
    AgentHandle,
    AgentResult,
    AgentTemplate,
    ResultMetadata,
    TaskSpec,
    TrustLevel,
)

Agent

Agent

A Murmur agent — frozen, broker-safe, serializable.

The model, instructions, output_type, and tools fields drive PydanticAI internally. The trust_level, context_passer, and backend fields drive Murmur orchestration. Users compose them on a single object; the runtime splits them apart at dispatch time.

name instance-attribute

name: str

Stable identifier used as the registry key, broker topic suffix, and agent_name field on every log line and RuntimeEvent.

model instance-attribute

model: str | Model

The agent's model — either a PydanticAI string identifier or a constructed :class:pydantic_ai.models.Model instance.

String form: "<provider>:<model_name>" — e.g. "anthropic:claude-sonnet-4-6", "openai:gpt-5.2", "openrouter:anthropic/claude-sonnet-4-5". PydanticAI auto-resolves the provider, applies the default authentication (env var per vendor), and constructs the matching :class:Model internally. Use this form for the common case.

Instance form: a constructed Model — used when you need a non-default Provider (Azure OpenAI, Bedrock-hosted Anthropic, Vertex Gemini), a custom HTTP client, custom auth, or a private base URL:

from murmur.models import OpenRouterModel from murmur.providers import OpenRouterProvider Agent( ... name="researcher", ... model=OpenRouterModel( ... "anthropic/claude-sonnet-4-5", ... provider=OpenRouterProvider(api_key="sk-or-..."), ... ), ... ..., ... )

The full Model and Provider matrix is re-exported from :mod:murmur.models and :mod:murmur.providers so user code never has to import from :mod:pydantic_ai directly.

Forwarded to PydanticAI verbatim at dispatch; Murmur does not maintain its own model registry.

fallback_models class-attribute instance-attribute

fallback_models: tuple[str, ...] = ()

Ordered fallback model names. () (default) means no fallbacks.

When non-empty, the runtime builds :class:pydantic_ai.models.fallback.FallbackModel(model, *fallback_models) at dispatch and uses it instead of model directly. The default fallback trigger is :class:pydantic_ai.ModelAPIError (4xx / 5xx) — the common "provider down / rate limited" case. Each entry is a PydanticAI-style model string ("openai:gpt-5.2", "google:gemini-3-pro-preview", etc.); per-fallback ModelSettings and Provider overrides are deferred (single model_settings is shared across primary + all fallbacks for now).

Agent( ... name="r", ... model="anthropic:claude-sonnet-4-6", ... fallback_models=("openai:gpt-5.2",), ... ..., ... )

Caveats:

• PydanticAI provider SDKs may have built-in retry logic that delays fallback activation. Set max_retries=0 on a custom client if you need immediate fallback.
• All-models-failed raises :class:pydantic_ai.FallbackExceptionGroup (an :class:ExceptionGroup subclass). User code that catches :class:pydantic_ai.ModelAPIError needs except* on Python 3.11+ to catch through the group; Murmur's :class:SpawnError translation at the dispatch boundary unwraps and stringifies whichever exception surfaces first, so most callers don't see the group.
• Validation errors (structured-output retries) do not trigger fallback — they use PydanticAI's per-model retry mechanism.

instructions instance-attribute

instructions: str

System prompt forwarded to PydanticAI as the agent's system_prompt. Plain string — variable interpolation happens upstream of construction (e.g. before the YAML loader resolves the spec).

output_type instance-attribute

output_type: type[BaseModel]

Pydantic model class the agent's output is validated against. PydanticAI re-prompts on validation failure up to its built-in retry budget; the runtime surfaces a final failure as :class:SpawnError.

input_type class-attribute instance-attribute

input_type: type[BaseModel] | None = None

Optional structured input type. None = the agent takes a plain string.

tools class-attribute instance-attribute

tools: frozenset[str] = Field(default_factory=frozenset)

Native tool names registered in the runtime's :class:ToolRegistry. Each call flows through :class:ToolExecutor for trust gating, allow-list filtering, and TOOL_CALL_* lifecycle events. Frozen — the agent's tool set is fixed at construction; use :meth:with_ to derive a variant.

mcp_servers class-attribute instance-attribute

mcp_servers: tuple[ToolsetProvider, ...] = ()

Remote toolset providers — tools discovered at dispatch time.

Each provider's tools are exposed to the agent alongside its native tools=… set. Calls flow through the same :class:ToolExecutor gate, so trust gating and lifecycle events apply identically. Build via :func:murmur.tools.mcp_stdio / :func:murmur.tools.mcp_http / :func:murmur.tools.mcp_sse.

The runtime owns provider lifecycle — it calls start() lazily on first dispatch and stop() on shutdown.

builtin_tools class-attribute instance-attribute

builtin_tools: tuple[AbstractBuiltinTool, ...] = ()

Provider-side built-in tools — executed by the LLM provider, not Murmur.

Examples: :class:pydantic_ai.WebSearchTool, :class:pydantic_ai.CodeExecutionTool, :class:pydantic_ai.ImageGenerationTool, :class:pydantic_ai.WebFetchTool, :class:pydantic_ai.FileSearchTool. Pass instances (with their own configuration knobs — max_uses, allowed_domains, etc.) in this tuple and they're forwarded to pydantic_ai.Agent(builtin_tools=...) at dispatch.

For ergonomics, the concrete classes are also re-exported from :mod:murmur.tools so users can avoid importing PydanticAI directly:

from murmur.tools import WebSearchTool Agent(name="r", model="anthropic:claude-sonnet-4-6", ... builtin_tools=(WebSearchTool(max_uses=5),), ...)

CAVEAT — these run on the provider's infrastructure, so they bypass Murmur's :class:ToolExecutor: no trust gate, no allow-list filtering, no per-tool TOOL_CALL_* lifecycle events (PydanticAI surfaces them post-hoc via ModelResponse.builtin_tool_calls). Token cost still flows through CostTrackingMiddleware because PydanticAI's usage() includes provider-side tool tokens. Provider support varies by tool — an unsupported combo raises UserError at run time, which surfaces as :class:SpawnError.

max_concurrent_requests class-attribute instance-attribute

max_concurrent_requests: int | None = None

Convenience cap on concurrent HTTP requests to this model.

When set to a positive integer, the runtime wraps the resolved model in :class:pydantic_ai.models.concurrency.ConcurrencyLimitedModel with a fresh per-agent :class:murmur.models.ConcurrencyLimiter. Distinct from :meth:AgentRuntime.gather's max_concurrency (which caps Murmur's task fan-out): this caps the provider-side request count, useful when a shared API key is rate-limited.

Agent(name="r", model="openai:gpt-5.2", max_concurrent_requests=5, ...)

Mutually exclusive with :attr:model_concurrency_limiter. Use the limiter field instead when several agents need to share one cap (e.g. one Anthropic key behind a fleet of agents).

model_concurrency_limiter class-attribute instance-attribute

model_concurrency_limiter: AbstractConcurrencyLimiter | None = None

Pre-built concurrency limiter shared across agents — wraps the resolved model with the same limiter instance every dispatch.

Build via :class:murmur.models.ConcurrencyLimiter (or any :class:murmur.models.AbstractConcurrencyLimiter subclass for custom backends — e.g. a Redis-backed cross-process limiter):

from murmur.models import ConcurrencyLimiter pool = ConcurrencyLimiter(max_running=10, name="openai-pool") head = Agent(name="head", model="openai:gpt-5.2", ... model_concurrency_limiter=pool, ...) minion = Agent(name="minion", model="openai:gpt-5.2", ... model_concurrency_limiter=pool, ...)

Mutually exclusive with :attr:max_concurrent_requests. Limiting is single-process by default — pass a custom AbstractConcurrencyLimiter subclass (e.g. Redis-backed) for cross-process limiting across a worker fleet.

model_settings class-attribute instance-attribute

model_settings: Mapping[str, object] | None = None

Per-provider knobs forwarded to the underlying model — temperature, max_tokens, top_p, etc.

The map is passed through to pydantic_ai.Agent(model_settings=...) verbatim. Recognised keys are provider-specific (PydanticAI validates per-provider at request time); a typo is a silent no-op rather than a Murmur error. Common keys:

• temperature: float
• max_tokens: int
• top_p: float
• stop_sequences: list[str]
• timeout: float — per-request, distinct from RuntimeOptions.timeout_seconds which gates the whole agent run

None (default) means PydanticAI picks per-provider defaults.

trust_level class-attribute instance-attribute

trust_level: TrustLevel = MEDIUM

Tool-access policy. Drives :class:ToolExecutor's gate (allow-list for LOW, full set for MEDIUM/HIGH, no tools for SANDBOX) and — once Phase 4 lands — backend selection (SANDBOX agents always run via ContainerBackend).

context_passer class-attribute instance-attribute

context_passer: ContextPasser = Field(default_factory=NullContextPasser)

Policy deciding what conversation history flows into a spawn. :class:NullContextPasser (default) hands the agent a fresh context; :class:FullContextPasser forwards everything. Phase 3 adds SummaryContextPasser and SelectiveContextPasser.

backend class-attribute instance-attribute

backend: str = 'auto'

Routing hint for :class:AgentRuntime to pick a :class:Backend. "auto" (default) defers to the runtime's configured backend (typically AsyncBackend in local mode, JobBackend when a broker URL was supplied). Reserved for future overrides — currently informational.

pre_process class-attribute instance-attribute

pre_process: tuple[ProcessHook, ...] = ()

Hooks applied left-to-right to the input before the LLM call.

Each hook is (input_type) -> input_type. Sync, pure — no I/O, no async. Empty tuple = identity.

post_process class-attribute instance-attribute

post_process: tuple[ProcessHook, ...] = ()

Hooks applied left-to-right to the output after the LLM call.

Each hook is (output_type) -> output_type. Sync, pure — no I/O, no async. Empty tuple = identity.

with_

with_(**updates: object) -> Agent

Return a copy with the given fields replaced — the only mutation path.

Source code in src/murmur/agent.py

def with_(self, **updates: object) -> Agent:
    """Return a copy with the given fields replaced — the only mutation path."""
    return self.model_copy(update=updates)

AgentTemplate

AgentTemplate

Frozen builder for shared :class:murmur.Agent config.

Materialize concrete agents via :meth:agent. The template itself is broker-safe — it serialises through Pydantic with no callables on its surface.

pre_instruction class-attribute instance-attribute

pre_instruction: str | None = None

Prepended to every materialised agent's instructions with a blank line between. None (default) means no prefix.

Useful for fleet-wide preamble: "You are part of the Murmur swarm. Always return JSON. Never apologize."

model class-attribute instance-attribute

model: str | Model | None = None

Default model for materialised agents — either a PydanticAI string identifier ("anthropic:claude-sonnet-4-6") or a constructed :class:pydantic_ai.models.Model instance (for non-default Provider / custom base URL — Azure, Bedrock, OpenRouter, LM Studio, Ollama, vLLM, etc.). Mirrors :attr:Agent.model. None (default) means the per-call kwarg must supply model=.

fallback_models class-attribute instance-attribute

fallback_models: tuple[str, ...] | None = None

Default fallback chain. See :attr:Agent.fallback_models.

input_type class-attribute instance-attribute

input_type: type[BaseModel] | None = None

Default input_type. See :attr:Agent.input_type.

tools class-attribute instance-attribute

tools: frozenset[str] | None = None

Default native tool set. Per-call tools= replaces this — it doesn't extend. Build a union explicitly when you want both.

mcp_servers class-attribute instance-attribute

mcp_servers: tuple[ToolsetProvider, ...] | None = None

Default MCP toolset providers. Per-call mcp_servers= replaces.

builtin_tools class-attribute instance-attribute

builtin_tools: tuple[AbstractBuiltinTool, ...] | None = None

Default provider-side built-in tools. Per-call replaces.

max_concurrent_requests class-attribute instance-attribute

max_concurrent_requests: int | None = None

Default per-agent provider HTTP concurrency cap. Mutually exclusive with :attr:model_concurrency_limiter on the template.

model_concurrency_limiter class-attribute instance-attribute

model_concurrency_limiter: AbstractConcurrencyLimiter | None = None

Default shared concurrency limiter. Mutually exclusive with :attr:max_concurrent_requests on the template.

model_settings class-attribute instance-attribute

model_settings: Mapping[str, object] | None = None

Default provider model_settings. See :attr:Agent.model_settings.

trust_level class-attribute instance-attribute

trust_level: TrustLevel | None = None

Default trust level. None means materialised agents fall back to :attr:Agent.trust_level's default (MEDIUM).

context_passer class-attribute instance-attribute

context_passer: ContextPasser | None = None

Default :class:ContextPasser. None means materialised agents fall back to :class:NullContextPasser (Agent's default).

agent

agent(
    *,
    name: str,
    instructions: str,
    output_type: type[BaseModel],
    model: str | Model | None = None,
    fallback_models: tuple[str, ...] | None = None,
    input_type: type[BaseModel] | None = None,
    tools: frozenset[str] | None = None,
    mcp_servers: tuple[ToolsetProvider, ...] | None = None,
    builtin_tools: tuple[AbstractBuiltinTool, ...] | None = None,
    max_concurrent_requests: int | None = None,
    model_concurrency_limiter: AbstractConcurrencyLimiter | None = None,
    model_settings: Mapping[str, object] | None = None,
    trust_level: TrustLevel | None = None,
    context_passer: ContextPasser | None = None,
    pre_process: tuple[ProcessHook, ...] = (),
    post_process: tuple[ProcessHook, ...] = (),
    backend: str = "auto",
) -> Agent

Materialize a concrete :class:Agent from this template.

name, instructions, and output_type are always per-agent. Every other kwarg is an optional override of the template's corresponding field; pass None (the default) to inherit.

pre_instruction (when set on the template) prefixes the per-agent instructions with a blank line between.

Source code in src/murmur/templates.py

def agent(
    self,
    *,
    name: str,
    instructions: str,
    output_type: type[BaseModel],
    model: str | Model | None = None,
    fallback_models: tuple[str, ...] | None = None,
    input_type: type[BaseModel] | None = None,
    tools: frozenset[str] | None = None,
    mcp_servers: tuple[ToolsetProvider, ...] | None = None,
    builtin_tools: tuple[AbstractBuiltinTool, ...] | None = None,
    max_concurrent_requests: int | None = None,
    model_concurrency_limiter: AbstractConcurrencyLimiter | None = None,
    model_settings: Mapping[str, object] | None = None,
    trust_level: TrustLevel | None = None,
    context_passer: ContextPasser | None = None,
    pre_process: tuple[ProcessHook, ...] = (),
    post_process: tuple[ProcessHook, ...] = (),
    backend: str = "auto",
) -> Agent:
    """Materialize a concrete :class:`Agent` from this template.

    ``name``, ``instructions``, and ``output_type`` are always per-agent.
    Every other kwarg is an optional override of the template's
    corresponding field; pass ``None`` (the default) to inherit.

    ``pre_instruction`` (when set on the template) prefixes the
    per-agent ``instructions`` with a blank line between.
    """
    if self.pre_instruction is not None:
        final_instructions = f"{self.pre_instruction}\n\n{instructions}"
    else:
        final_instructions = instructions

    kwargs: dict[str, Any] = {
        "name": name,
        "instructions": final_instructions,
        "output_type": output_type,
        "pre_process": pre_process,
        "post_process": post_process,
        "backend": backend,
    }

    # For each templatable field: per-call override wins; else template
    # value if set; else omit so Agent's own default applies.
    per_field: tuple[tuple[str, object, object], ...] = (
        ("model", model, self.model),
        ("fallback_models", fallback_models, self.fallback_models),
        ("input_type", input_type, self.input_type),
        ("tools", tools, self.tools),
        ("mcp_servers", mcp_servers, self.mcp_servers),
        ("builtin_tools", builtin_tools, self.builtin_tools),
        (
            "max_concurrent_requests",
            max_concurrent_requests,
            self.max_concurrent_requests,
        ),
        (
            "model_concurrency_limiter",
            model_concurrency_limiter,
            self.model_concurrency_limiter,
        ),
        ("model_settings", model_settings, self.model_settings),
        ("trust_level", trust_level, self.trust_level),
        ("context_passer", context_passer, self.context_passer),
    )
    for field_name, call_value, template_value in per_field:
        resolved = call_value if call_value is not None else template_value
        if resolved is not None:
            kwargs[field_name] = resolved

    return Agent(**kwargs)

TaskSpec

TaskSpec

A single unit of work handed to runtime.run / runtime.gather.

id class-attribute instance-attribute

id: str = Field(default_factory=lambda: str(uuid4()))

Per-task UUID. Auto-generated; collisions on the broker results topic are correlated by this value.

request_id class-attribute instance-attribute

request_id: str = Field(default_factory=lambda: str(uuid4()))

Correlates one logical request across logs / broker messages / HTTP.

Generated per task by default; supply explicitly to thread an upstream id (e.g. an X-Request-Id header) through every layer of the runtime.

input instance-attribute

input: str

The agent's input — a plain string, or a JSON-serialised structure when Agent.input_type is set (the runtime decodes against the agent's declared input type at dispatch).

metadata class-attribute instance-attribute

metadata: Mapping[str, str] = Field(default_factory=dict)

Free-form string-to-string metadata. Surfaces on every emitted RuntimeEvent and on the broker wire envelope; use for tenant / customer / trace tags. Frozen at construction.

AgentResult

AgentResult

The typed envelope every runtime.run call returns.

Either output is set (success) or error is set (failure) — never both. Use :meth:is_ok to discriminate.

output class-attribute instance-attribute

output: T | None = None

The agent's structured output, validated against Agent.output_type. None when error is set.

error class-attribute instance-attribute

error: BaseException | None = None

The exception that caused the run to fail. None on success. Always a :class:MurmurError subclass when raised by the runtime; user-tool exceptions wrap in :class:ToolExecutionError.

metadata class-attribute instance-attribute

metadata: ResultMetadata = Field(default_factory=ResultMetadata)

Per-result diagnostics — duration, tokens, cost, backend, trace_id.

agent_name instance-attribute

agent_name: str

Name of the :class:Agent that produced this result. Mirrors Agent.name.

task_id instance-attribute

task_id: str

The originating TaskSpec.id — correlates a result back to its request.

is_ok

is_ok() -> bool

True if the agent succeeded and output is populated.

Source code in src/murmur/types.py

def is_ok(self) -> bool:
    """``True`` if the agent succeeded and ``output`` is populated."""
    return self.error is None and self.output is not None

GroupResult

GroupResult

Multi-leaf result from :meth:AgentRuntime.run_group.

Returned when an :class:AgentGroup topology has more than one terminal node fire — typically a moderator-and-specialists shape where each specialist is its own leaf rather than feeding a single synthesiser. Single-leaf topologies still return a plain :class:AgentResult for backward compatibility.

Iteration: GroupResult.outputs is keyed by Agent.name so callers can pick out a specific terminal by name. Use :attr:terminal for the single-leaf convenience case.

outputs instance-attribute

outputs: Mapping[str, AgentResult[BaseModel]]

Per-leaf results keyed by Agent.name. A leaf that was skipped at runtime (branch routing condition, heterogeneous fan-out filter empty) is absent from this mapping — present keys correspond to terminals that actually fired.

Insulated from caller mutation by an input-copy in the after-validator: the dict the model stores is independent of whatever the constructor was handed. Pydantic's model_config(frozen=True) blocks whole-attribute reassignment (result.outputs = ...); reaching through the stored dict reference (result.outputs["new"] = ...) is technically possible but undefined behaviour — treat GroupResult as read-only after construction.

metadata class-attribute instance-attribute

metadata: ResultMetadata = Field(default_factory=ResultMetadata)

Aggregate diagnostics across every fired leaf. tokens_used sums; duration_ms takes the max (parallel tiers don't add durations); cost_usd sums; backend is the literal string "group"; trace_id mirrors the task's request_id.

terminal property

terminal: AgentResult[BaseModel]

Convenience accessor for single-leaf cases.

Raises :class:ValueError when the group fired more than one terminal — callers must use the keyed outputs mapping in that case.

AgentHandle

AgentHandle

Opaque handle returned by a backend's spawn — used to kill / await.

handle_id class-attribute instance-attribute

handle_id: str = Field(default_factory=lambda: str(uuid4()))

Backend-issued UUID. Treated as opaque by callers.

agent_name instance-attribute

agent_name: str

Mirrors Agent.name for the spawned run.

task_id instance-attribute

task_id: str

Mirrors TaskSpec.id for the dispatched task.

backend instance-attribute

backend: str

Class name of the :class:Backend that issued this handle.

AgentContext

AgentContext

Context object passed between stages in the pipeline.

Carries the conversation history, parent agent reference (for cascading spawns), and any user-attached metadata. Stages may produce a new AgentContext via model_copy(update=...) but never mutate the one they receive.

messages class-attribute instance-attribute

messages: tuple[Mapping[str, str], ...] = Field(default_factory=tuple)

Conversation history forwarded into the spawn. Each entry is a {"role": "user"|"assistant"|"system", "content": str} mapping. Empty tuple = fresh context. The :class:ContextPasser chosen on the agent decides what fills this on each spawn.

parent_agent class-attribute instance-attribute

parent_agent: str | None = None

Name of the immediate parent agent, when this is a sub-spawn. None for top-level runs.

parent_trace_id class-attribute instance-attribute

parent_trace_id: str | None = None

trace_id of the parent run that issued this sub-spawn. Threaded through onto every child :class:RuntimeEvent so observability backends can stitch a cascading run into a single tree. None for top-level runs.

ancestors class-attribute instance-attribute

ancestors: frozenset[str] = Field(default_factory=frozenset)

Set of agent names currently above this run in the spawn chain. Empty for top-level runs; for a sub-spawn it contains every ancestor up to the top-level agent. The runtime rejects a spawn whose target name already appears in ancestors with :class:SpawnCycleError — preventing A → B → A reentry without a separate graph store.

depth class-attribute instance-attribute

depth: int = 0

Cascading-spawn depth. 0 for top-level runs; incremented per sub-spawn. :class:DepthLimitMiddleware rejects when this reaches RuntimeOptions.max_spawn_depth.

metadata class-attribute instance-attribute

metadata: Mapping[str, str] = Field(default_factory=dict)

Free-form context metadata, threaded through to the spawned agent. Distinct from TaskSpec.metadata — that's per-task; this is per-context.

ResultMetadata

ResultMetadata

Per-result diagnostics produced by the runtime.

duration_ms class-attribute instance-attribute

duration_ms: int = 0

Wall-clock time from spawn to result, in milliseconds.

tokens_used class-attribute instance-attribute

tokens_used: int = 0

Total tokens consumed (request + response + provider-side built-in tool tokens). Driver behind :class:CostTrackingMiddleware's post-charge.

cost_usd class-attribute instance-attribute

cost_usd: float = 0.0

Best-effort USD cost computed from tokens_used and the model's published rates. 0.0 when rates aren't known for the model in use.

backend class-attribute instance-attribute

backend: str = ''

Class name of the :class:Backend that ran the agent (e.g. "AsyncBackend", "JobBackend"). Empty until populated by the backend's result path.

trace_id class-attribute instance-attribute

trace_id: str | None = None

Same value as TaskSpec.request_id for the run that produced this result — populated when available. None for synthetic results.

TrustLevel

TrustLevel

Tool-access policy applied to an agent at runtime.

HIGH class-attribute instance-attribute

HIGH = 'high'

Full tool access.

MEDIUM class-attribute instance-attribute

MEDIUM = 'medium'

Curated tool set — the default for most agents.

LOW class-attribute instance-attribute

LOW = 'low'

Read-only tools only.

SANDBOX class-attribute instance-attribute

SANDBOX = 'sandbox'

No tools — pure reasoning.