symfonic

symfonic

symfonic — the canonical entry point.

from symfonic import Agent

agent = Agent(AnthropicProvider())
result = await agent.run("Say hello.")

That is the whole of the simple API: one required dependency, no FrameworkConfig, no tenant scope, no context manager, no background flush, and nothing printed about capabilities you did not ask for.

Everything here is resolved through a PEP 562 module __getattr__ and the module body executes no package import (FAC-1). Two consequences, both deliberate:

• import symfonic.memory costs exactly what it cost before this module existed — the top level taxes nobody.
• from symfonic import Agent imports only the facade chain, which is what makes the import-hermeticity contract (QUI-1: no symfonic.memory*, no fastapi, no mcp, …) mechanically checkable rather than aspirational.

__all__ is exactly the twelve names in T2.1.1's facade-surface.json. A thirteenth needs a preservation-matrix decision (API-ADR §2 A3) and a deprecation-registry row, which is the point of writing it down here.

Stability. Every new name enters at tier 2 (provisional), per API-ADR §1's rule for new surfaces: the import path is already stable, the shape may still change at a minor with a registry entry, and promotion to tier 1 is gated on W2 exit. Attachment and StructuredOutputError are tier-1 classes whose canonical paths are unchanged — the top-level alias is an additional path, never a move, so nothing here deprecates anything.

From 11.3, constructing SymfonicAgent or AgentBuilder raises EntryPointRemovedError (CORE-388). Use symfonic.Agent for one agent, symfonic.platform.AgentHost to own shared resources and per-scope agents in a process, and symfonic.migration.agent_from_config to carry an existing FrameworkConfig onto an Agent; it refuses, by name, every active setting it does not translate.

Agent

Agent(model_provider: Any, *, instructions: str | None = None, model: str | ModelConfig | None = None, tools: Sequence[Any] = (), result_policy: ResultPolicySettings | None = None, capabilities: Sequence[CapabilityConfig] = (), max_model_rounds: int | None = None, observe_provider_call: Callable[[object], None] | None = None)

Bases: AgentContinuationMixin, AgentInstrumentationMixin

A tool-capable agent with an optional receipt plan cache.

Parameters:

Name	Type	Description	Default
model_provider	Any	Any object satisfying symfonic.core.ModelProvider — AnthropicProvider, OpenAIProvider, MockModelProvider, or an adopter's own class. The sole required dependency.	required
instructions	str | None	The system prompt, used verbatim. None sends no system message at all; the framework substitutes no default.	None
tools	Sequence[Any]	A @symfonic_tool object, a LangChain BaseTool, or a plain annotated callable — fixed at construction.	()
result_policy	ResultPolicySettings | None	Explicit native tool-result policy settings.	None
capabilities	Sequence[CapabilityConfig]	Typed optional behavior; grants are separated before folding.	()
max_model_rounds	int | None	Provider round limit; None keeps the facade default.	None
observe_provider_call	Callable[[object], None] | None	Optional local observer of a deep copy of each model-call message argument. Disabled by default; never exported.	None

Construction has no effects (FAC-8); the first effect occurs in run().

Ordinary instances serve concurrent run/stream calls on one loop; receipt instances cache one plan and the journal arbitrates claims (FAC-9). It is not documented as safe to share across event loops.

Source code in src/symfonic/agent/facade.py

def __init__(
    self,
    model_provider: Any,
    *,
    instructions: str | None = None,
    model: str | ModelConfig | None = None,
    tools: Sequence[Any] = (),
    result_policy: ResultPolicySettings | None = None,
    capabilities: Sequence[CapabilityConfig] = (),
    max_model_rounds: int | None = None,
    observe_provider_call: Callable[[object], None] | None = None,
) -> None:
    require_provider(model_provider)
    require_instructions(instructions)
    resolved_model = as_model_config(model)
    require_round_budget(max_model_rounds)
    require_native_result_policy(result_policy)
    # GrantEffects is authority, not a capability; remove it before folding.
    adopter_grants, capabilities = partition_grants(tuple(capabilities))
    receipt, capabilities = require_capabilities(capabilities)
    if receipt is not None and observe_provider_call is not None:
        raise ConfigurationError("receipt Agent cannot bind model-call observers")
    self._closed, self._receipt_plan_cache = False, ({} if receipt else None)
    sinks = event_sink_bindings(capabilities)
    if len(sinks) > 1:
        raise ConfigurationError(
            "capabilities contributed more than one event sink; one run "
            "has one ordered observer stream."
        )
    self._event_binding = sinks[0] if sinks else None
    baseline_grants = frozenset({"model_call"}) | (
        frozenset({"tool_call"}) if tools or (result_policy and result_policy.enabled)
        else frozenset()
    )
    authorized_effects = baseline_grants | adopter_grants
    cache_composition = cache_composition_for(capabilities)
    folded = fold_contributions(tuple(capabilities), effect_grants=authorized_effects)
    stages, handlers, _grants, capability_tools, names, preconditions = folded
    self._capabilities = tuple(names)
    self.__composed_capabilities = tuple(capabilities)
    self._composition_manifest = composition_manifest(
        stages, capability_tools, names, preconditions
    )
    # One normalised set feeds binding, manifest, execution and grants (#31).
    if result_policy is not None and result_policy.enabled:
        from symfonic.agent.backend.result_recall import recall_tool_result
        capability_tools = (*capability_tools, ("result-policy", recall_tool_result))
    merged = merge_capability_tools(tools, capability_tools)
    self._plans = AgentPlanFactory(
        model_provider=model_provider,
        instructions=instructions,
        model=resolved_model,
        tools=merged,
        result_policy_settings=result_policy,
        **({} if max_model_rounds is None else {"max_model_rounds": max_model_rounds}),
        stages=stages,
        stage_handlers=handlers,
        # Never collected into a grant: a precondition can only refuse.
        tool_preconditions=preconditions,
        cache_composition=cache_composition,
        terminal_synthesis=terminal_synthesis_for(capabilities),
        # Not derived from contributions: dispatch needs baseline authority too.
        authorized_effects=authorized_effects,
        capability_names=names,
        observe_provider_call=observe_provider_call,
    )

capabilities property

capabilities: tuple[str, ...]

Names actually folded into this agent's invocation plan.

composition_manifest property

composition_manifest: Mapping[str, Any]

Payload-free attestation of the capability fold this agent retained.

close async

close() -> None

Release facade-owned resources. Idempotent, and never raises.

In W1 the facade owns nothing, so this only marks the instance closed. It never closes anything the adopter passed in — the provider outlives the agent. An in-flight run() is allowed to finish; close() does not cancel it (LIF-4).

Source code in src/symfonic/agent/facade.py

async def close(self) -> None:
    """Release facade-owned resources. Idempotent, and never raises.

    In W1 the facade owns nothing, so this only marks the instance closed.
    It never closes anything the adopter passed in — the provider outlives
    the agent. An in-flight ``run()`` is allowed to finish; ``close()``
    does not cancel it (LIF-4).
    """
    self._closed = True

run async

run(prompt: str, *, attachments: Sequence[Attachment] = ..., history: Sequence[Message] = ..., state: Mapping[str, Any] | None = ..., session_id: str = ..., invocation: AgentInvocation | None = ...) -> AgentResult[None]

run(prompt: str, *, attachments: Sequence[Attachment] = ..., history: Sequence[Message] = ..., state: Mapping[str, Any] | None = ..., session_id: str = ..., invocation: AgentInvocation | None = ..., output_type: type[OutputT]) -> AgentResult[OutputT]

run(prompt: str, *, attachments: Sequence[Attachment] = (), history: Sequence[Message] = (), state: Mapping[str, Any] | None = None, session_id: str = '', invocation: AgentInvocation | None = None, output_type: type[BaseModel] | None = None) -> AgentResult[Any]

Run one non-streaming turn.

history is how a stateless agent takes a second turn: the adopter holds the transcript and passes result.messages back.

Source code in src/symfonic/agent/facade.py

async def run(
    self,
    prompt: str,
    *,
    attachments: Sequence[Attachment] = (),
    history: Sequence[Message] = (),
    state: Mapping[str, Any] | None = None,
    session_id: str = "",
    invocation: AgentInvocation | None = None,
    output_type: type[BaseModel] | None = None,
) -> AgentResult[Any]:
    """Run one non-streaming turn.

    ``history`` is how a stateless agent takes a second turn: the adopter
    holds the transcript and passes ``result.messages`` back.
    """
    request = turn_for(prompt, attachments, history, state,
                       session_id=session_id, invocation=invocation)
    return await dispatch(self._kernel, self._prepare, request, output_type,
                          self._composed_capabilities)

stream

stream(prompt: str, *, attachments: Sequence[Attachment] = ..., history: Sequence[Message] = ..., state: Mapping[str, Any] | None = ..., session_id: str = ..., invocation: AgentInvocation | None = ...) -> AsyncIterator[AgentEvent]

stream(prompt: str, *, attachments: Sequence[Attachment] = ..., history: Sequence[Message] = ..., state: Mapping[str, Any] | None = ..., session_id: str = ..., invocation: AgentInvocation | None = ..., output_type: type[OutputT]) -> AsyncIterator[AgentEvent]

stream(prompt: str, *, attachments: Sequence[Attachment] = (), history: Sequence[Message] = (), state: Mapping[str, Any] | None = None, session_id: str = '', invocation: AgentInvocation | None = None, output_type: type[BaseModel] | None = None) -> AsyncIterator[AgentEvent]

Run one streaming turn.

A plain def, not an async generator function, on purpose: an async generator defers argument validation to the first __anext__, which would surface a ConfigurationError after the caller believed the stream had started. This validates eagerly and raises at the call site, while async for event in agent.stream(...) still reads identically (EVT-8).

Source code in src/symfonic/agent/facade.py

def stream(
    self,
    prompt: str,
    *,
    attachments: Sequence[Attachment] = (),
    history: Sequence[Message] = (),
    state: Mapping[str, Any] | None = None,
    session_id: str = "",
    invocation: AgentInvocation | None = None,
    output_type: type[BaseModel] | None = None,
) -> AsyncIterator[AgentEvent]:
    """Run one streaming turn.

    A plain ``def``, not an async generator function, on purpose: an async
    generator defers argument validation to the first ``__anext__``, which
    would surface a ``ConfigurationError`` *after* the caller believed the
    stream had started. This validates eagerly and raises at the call site,
    while ``async for event in agent.stream(...)`` still reads identically
    (EVT-8).
    """
    request = turn_for(prompt, attachments, history, state,
                       session_id=session_id, invocation=invocation)
    return stream_dispatch(self._kernel, self._prepare, request, output_type,
                           self._composed_capabilities)

AgentEvent

Bases: BaseModel

One event of an Agent.stream() (EVT-1…EVT-10).

A projection of the same invocation the non-streaming path runs — not a second pipeline. Field population per kind:

============== =================================================== kind populated fields ============== =================================================== thinking text (reasoning delta) text_delta text (answer delta) tool_call tool_call with result/error both None tool_result tool_call with result or error set done result error error cancelled (none) ask_user interrupt interrupt interrupt stage stage, and stage.counts where it counted ============== ===================================================

stage is what a capability's execution reported: which stage ran, in which phase, for which capability, with what outcome and why. It shares the dense index with everything else, so a consumer can place a retrieval against the round it fed. It carries counts and never contents -- see :class:StageRecord.

AgentResult

Bases: BaseModel, Generic[OutputT]

The result of one Agent.run() (RES-1…RES-8).

text is never None — a run that produced no text yields "". AgentResponse.final_response's None-vs-"" tri-state has no meaning a caller can act on and forces a null check on the happiest path.

messages is the whole turn in order and is designed to be fed straight back as the next call's history; that round trip is how a stateless agent takes a second turn.

Attachment

Bases: BaseModel

Non-text content to send alongside a query.

The semantic query string is used for HMS hydration and embeddings; attachments are forwarded to the LLM as additional content blocks but do NOT participate in memory hydration or tool routing in v1.

Attributes:

Name	Type	Description
kind	Literal['image', 'document']	Content type -- "image" or "document".
source_type	Literal['url', 'base64']	How data is encoded -- "url" (http/https only -- other schemes such as file:// are rejected to prevent SSRF) or "base64".
data	str	URL string or base64-encoded content.
media_type	str	MIME type (e.g. "image/png", "application/pdf"). Defaults to image/png for kind="image" and application/pdf for kind="document".
filename	str | None	Optional filename for documents / PDFs.

CapabilityConfig

Bases: Protocol

Marker protocol for the API-ADR A5 optional-behavior seam (FAC-7).

Optional behavior attaches by passing a typed configuration object, and absence means disabled — there are no boolean feature flags on the facade, ever. W1 registers zero capabilities, so the parameter exists solely so the first real capability is an additive change rather than a signature change.

ConfigurationError

Bases: SymfonicError

Raised when capability flags are violated at compile time.

Promoted from symfonic.core.graph (FERR-2). The facade raises it for bad constructor input, an unregistrable capability, and a provider that cannot satisfy a requested structured output — always before the first provider call (FERR-3).

ContractViolationError

Bases: SymfonicError

Raised when a documented facade contract is misused.

W1's only use is LIF-4: run()/stream() after close(). Closed is terminal and there is no reopen, so continuing would be a silent lie about the agent's state.

A PRE_MODEL, POST_MODEL, PRE_TOOL, POST_TOOL or FINALIZE stage handler may preserve an adopter-defined refusal type across the :class:symfonic.Agent boundary by subclassing this error and setting preserve_contract_identity = True on the subclass. Without that explicit opt-in, a handler exception is wrapped in this base type and retained as its __cause__. PROMPT_ASSEMBLY always wraps, including opted-in subclasses, through its separate prompt-assembly failure guard.

Message

Bases: BaseModel

One typed text message; attachments are not replayed through history.

StructuredOutputError

StructuredOutputError(message: str)

Bases: SymfonicAgentError

Raised when a structured-output extraction cannot be satisfied.

Carries code="unprocessable" so the FastAPI layer maps it to a 422 rather than a generic 500 — the request was well-formed, but the model could not produce a value matching the requested schema.

Source code in src/symfonic/agent/structured.py

def __init__(self, message: str) -> None:
    super().__init__(message, code="unprocessable")

SymfonicError

Bases: Exception

Root of the simple-facade error taxonomy (FERR-1).

Never raised directly. It exists so an adopter can write one except SymfonicError and catch every failure the facade is documented to raise — and nothing else. Provider SDK exceptions deliberately do not inherit from it: W1 propagates them unchanged rather than shipping a half-built wrapper that swallows the provider's own diagnostics (FERR-4).

TokenUsage

Bases: BaseModel

What the provider reported (RES-4).

0 means "unreported", not "free": the facade never estimates token counts, so a provider that reports nothing leaves the zero value.

ToolCall

Bases: BaseModel

One tool invocation and its outcome (RES-3).

On a completed run exactly one of result / error is non-None. Both are None only on the in-flight tool_call stream event, which is what makes id the join key to the matching tool_result.