symfonic.agent.facade

facade

Capability-composed Agent: one provider, optional behavior as capabilities.

Implements T2.1.1's FAC-4…FAC-9 and LIF-1…LIF-5.

Agent

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

Bases: AgentContinuationMixin

A stateless, tool-capable agent.

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.	()
capabilities	Sequence[CapabilityConfig]	The typed optional-behavior seam. Absence means disabled; there are no boolean feature flags. Anything with a contribute(request) -> CapabilityContribution is accepted — MemoryCapability and PromptingCapability are the two that have one today. A GrantEffects in this sequence is removed before anything is asked to contribute, because a capability that could widen the grant set would be granting itself.	()
max_model_rounds	int | None	How many provider round trips one run() may take before the kernel stops with stop_reason="tool_limit". None keeps the facade default (MAX_TOOL_ITERATIONS). A caller migrating a tool-heavy agent off the legacy body passes the budget it already had, because a loop bounded lower than the path it replaced returns a partial answer as if it were whole.	None

Construction validates and stores, and does nothing else (FAC-8): no connection, no task, no file, no environment read, no chat model. The first side effect of an Agent is the provider call inside run(). That is what makes async with and close() optional (LIF-1) — an agent that acquired nothing leaks nothing.

One instance may serve unlimited concurrent run/stream calls on one event loop, because no per-invocation state is stored on it (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] = (),
    capabilities: Sequence[CapabilityConfig] = (),
    max_model_rounds: int | None = None,
) -> None:
    require_provider(model_provider)
    require_instructions(instructions)
    resolved_model = as_model_config(model)
    require_round_budget(max_model_rounds)
    # GrantEffects is authority, not a capability; remove it before folding.
    adopter_grants, capabilities = partition_grants(tuple(capabilities))
    for capability in capabilities:
        if not callable(getattr(capability, "contribute", None)):
            raise ConfigurationError(
                f"{type(capability).__name__} is not a CapabilityConfig: it "
                "has no contribute(request) -> CapabilityContribution. The "
                "seam takes a declaration plus the handlers that answer it, "
                "so a capability cannot register a stage nothing runs."
            )

    self._closed = False
    sinks = tuple(
        sink for capability in capabilities
        if (sink := getattr(capability, "event_sink", None)) is not None
    )
    if len(sinks) > 1:
        raise ConfigurationError(
            "capabilities contributed more than one event sink; one run "
            "has one ordered observer stream."
        )
    self._event_sink = sinks[0] if sinks else None
    # Baseline authority comes from the invocation shape, never capabilities.
    baseline_grants = frozenset({"model_call"}) | (
        frozenset({"tool_call"}) if tools else frozenset()
    )
    authorized_effects = baseline_grants | adopter_grants
    # ``grants`` is dropped: already checked, and carrying it on as
    # ``capability_grants`` once left a field nothing read.
    folded = fold_contributions(tuple(capabilities), effect_grants=authorized_effects)
    stages, handlers, _grants, capability_tools, names, preconditions = folded
    self._capabilities = tuple(names)
    self._composition_manifest = composition_manifest(
        stages, capability_tools, names, preconditions
    )
    # One normalised set feeds binding, manifest, execution and grants (#31).
    merged = merge_capability_tools(tools, capability_tools)
    self._plans = AgentPlanFactory(
        model_provider=model_provider,
        instructions=instructions,
        model=resolved_model,
        tools=merged,
        **({} 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,
        # Its own channel: not derived from contributions. Without it the
        # fold accepts memory-read and dispatch still refuses it -- green at
        # construction, dead at runtime.
        authorized_effects=authorized_effects,
        capability_names=names,
    )

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 = ...) -> AgentResult[None]

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

run(prompt: str, *, attachments: Sequence[Attachment] = (), history: Sequence[Message] = (), state: Mapping[str, Any] | None = None, session_id: str = '', 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 = "",
    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.
    """
    return await dispatch(
        self._kernel,
        self._prepare,
        turn_for(prompt, attachments, history, state, session_id=session_id),
        output_type,
    )

stream

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

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

stream(prompt: str, *, attachments: Sequence[Attachment] = (), history: Sequence[Message] = (), state: Mapping[str, Any] | None = None, session_id: str = '', 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 = "",
    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).
    """
    return stream_dispatch(
        self._kernel,
        self._prepare,
        turn_for(prompt, attachments, history, state, session_id=session_id),
        output_type,
    )