symfonic.agent.facade

facade

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

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)