`symfonic.capabilities.extensions`¶

extensions ¶

Extension composition: MCP servers and domain plugins as typed contributions.

Satisfies plan task T4.2.2. An MCP server and a domain plugin are the same thing from the framework's side — code the framework did not write, wanting to add to what the agent can do — and both now do it the same way: they declare tools, prompt fragments, policies, and lifecycle hooks, and the composer decides what survives.

Nothing here mutates an engine, patches a method, or registers into global state. :func:~.composition.compose is a pure function from contributions to one frozen value, which is what makes "installing this extension changed nothing else" an assertion a test can make.

Start at :mod:~symfonic.capabilities.extensions.contracts for the contract an extension author writes against.

ComposedExtensions `dataclass` ¶

ComposedExtensions(extensions: tuple[str, ...] = (), tools: tuple[ToolContribution, ...] = (), prompts: tuple[PromptFragment, ...] = (), policies: tuple[PolicyContribution, ...] = (), lifecycle: tuple[LifecycleContribution, ...] = (), diagnostics: tuple[ExtensionDiagnostic, ...] = ())

What a set of extensions adds up to, and what was refused on the way.

refusals `property` ¶

refusals: tuple[ExtensionDiagnostic, ...]

Only the error-severity diagnostics: what an operator must see.

decide `async` ¶

decide(request: PolicyRequest) -> PolicyVerdict

Combine every applicable policy's verdict, deny-wins.

Answers only the combined verdict. Use :meth:decide_detailed when the caller needs to know which guards failed to answer as well.

Source code in src/symfonic/capabilities/extensions/composition.py

async def decide(self, request: PolicyRequest) -> PolicyVerdict:
    """Combine every applicable policy's verdict, deny-wins.

    Answers only the combined verdict. Use :meth:`decide_detailed` when the
    caller needs to know *which* guards failed to answer as well.
    """
    return (await self.decide_detailed(request)).verdict

decide_detailed `async` ¶

decide_detailed(request: PolicyRequest) -> PolicyOutcome

Ask every applicable policy and report the verdict with the abstentions.

The first DENY short-circuits: once one policy has refused, asking the rest costs latency on a turn that is already refused, and no later verdict could change the answer. The abstentions reported are therefore those gathered before the refusal, which is the whole set whenever nothing denied.

A policy that raises abstains. That is the legacy engine's fail-open behaviour preserved on purpose — flipping it to fail-closed would turn a broken guard into a total outage — but the abstention is now a value the caller can see rather than a DEBUG log nobody reads.

Source code in src/symfonic/capabilities/extensions/composition.py

async def decide_detailed(self, request: PolicyRequest) -> PolicyOutcome:
    """Ask every applicable policy and report the verdict with the abstentions.

    The first DENY short-circuits: once one policy has refused, asking the
    rest costs latency on a turn that is already refused, and no later
    verdict could change the answer. The abstentions reported are therefore
    those gathered *before* the refusal, which is the whole set whenever
    nothing denied.

    A policy that raises abstains. That is the legacy engine's fail-open
    behaviour preserved on purpose — flipping it to fail-closed would turn
    a broken guard into a total outage — but the abstention is now a value
    the caller can see rather than a DEBUG log nobody reads.
    """
    abstentions: list[PolicyVerdict] = []
    for policy in self.policies:
        if not policy.applies_to(request.action):
            continue
        verdict = await self._ask(policy, request)
        if verdict.denied:
            return PolicyOutcome(verdict=verdict, abstentions=tuple(abstentions))
        if verdict.abstained:
            abstentions.append(verdict)
    return PolicyOutcome(
        verdict=PolicyVerdict.allow(reason="no contributed policy objected"),
        abstentions=tuple(abstentions),
    )

install `async` ¶

install() -> tuple[ExtensionDiagnostic, ...]

Run every install hook. A hook that raises costs a diagnostic.

Source code in src/symfonic/capabilities/extensions/composition.py

async def install(self) -> tuple[ExtensionDiagnostic, ...]:
    """Run every install hook. A hook that raises costs a diagnostic."""
    return await self._run_phase(LifecyclePhase.INSTALL)

teardown `async` ¶

teardown() -> tuple[ExtensionDiagnostic, ...]

Run every teardown hook, in reverse declaration order.

Reverse because unwinding is the inverse of building: an extension installed after another may depend on it, and closing in declaration order would tear the dependency out first.

Source code in src/symfonic/capabilities/extensions/composition.py

async def teardown(self) -> tuple[ExtensionDiagnostic, ...]:
    """Run every teardown hook, in reverse declaration order.

    Reverse because unwinding is the inverse of building: an extension
    installed after another may depend on it, and closing in declaration
    order would tear the dependency out first.
    """
    return await self._run_phase(LifecyclePhase.TEARDOWN, reverse=True)

ContributionKind ¶

Bases: StrEnum

The four things an extension may contribute, and there is no fifth.

ExtensionContractError ¶

Bases: ExtensionError

A contribution's own declaration is malformed.

Raised at declaration time — when the contribution value is validated — not at use time. A tool with no name is not a tool that fails when called; it is a tool that should never have been admitted to the catalogue.

ExtensionContribution `dataclass` ¶

ExtensionContribution(extension: str, tools: tuple[ToolContribution, ...] = (), prompts: tuple[PromptFragment, ...] = (), policies: tuple[PolicyContribution, ...] = (), lifecycle: tuple[LifecycleContribution, ...] = ())

Everything one extension offers, in one value.

An extension is asked, it answers, and the answer is validated, ordered, and refused as a unit. Nothing here holds a reference to an engine, a registry, or a graph, which is why an extension cannot install itself.

build `classmethod` ¶

build(extension: str, *, tools: Sequence[ToolContribution] = (), prompts: Sequence[PromptFragment] = (), policies: Sequence[PolicyContribution] = (), lifecycle: Sequence[LifecycleContribution] = ()) -> ExtensionContribution

Build a bundle from any sequences, freezing each into a tuple.

Source code in src/symfonic/capabilities/extensions/contracts.py

@classmethod
def build(
    cls,
    extension: str,
    *,
    tools: Sequence[ToolContribution] = (),
    prompts: Sequence[PromptFragment] = (),
    policies: Sequence[PolicyContribution] = (),
    lifecycle: Sequence[LifecycleContribution] = (),
) -> ExtensionContribution:
    """Build a bundle from any sequences, freezing each into a tuple."""
    return cls(
        extension=extension,
        tools=tuple(tools),
        prompts=tuple(prompts),
        policies=tuple(policies),
        lifecycle=tuple(lifecycle),
    )

validate ¶

validate() -> None

Validate the bundle, every member, and the attribution of each.

Attribution is checked here rather than on each member because it is a property of the pair: a well-formed tool contributed under someone else's name is exactly the confused-deputy shape AS-INT-5 is about, and the member alone cannot see that it was misfiled.

Source code in src/symfonic/capabilities/extensions/contracts.py

def validate(self) -> None:
    """Validate the bundle, every member, and the attribution of each.

    Attribution is checked here rather than on each member because it is a
    property of the *pair*: a well-formed tool contributed under someone
    else's name is exactly the confused-deputy shape AS-INT-5 is about, and
    the member alone cannot see that it was misfiled.
    """
    if not self.extension:
        raise ExtensionContractError("an extension contribution must name itself.")
    require_id(self.extension, what="extension", owner=self.extension)
    for group in (self.tools, self.prompts, self.policies, self.lifecycle):
        for member in group:
            member.validate()
            if member.extension != self.extension:
                raise PrivilegeWideningError(
                    f"{self.extension}: contributed a member attributed to "
                    f"{member.extension!r}. An extension may not contribute on "
                    "another extension's behalf."
                )

ExtensionDiagnostic `dataclass` ¶

ExtensionDiagnostic(extension: str, kind: ContributionKind, detail: str, subject: str = '', severity: Severity = Severity.WARNING)

One thing the composer decided, in names, for the record.

Diagnostics are how a dropped contribution stays visible. Nothing is discarded silently: every refusal, truncation, and remap emits one of these, and :class:~.composition.ComposedExtensions carries them alongside what survived.

ExtensionError ¶

Bases: ConfigurationError

Root of the extension-composition taxonomy.

Never raised directly. It exists so a composition root can write one except ExtensionError around the whole compose step and know it caught every way an extension can refuse to be composed.

ExtensionProvider ¶

Bases: Protocol

What the composer asks. Two members, both deliberate.

name is a declared member so an isinstance check requires it: an anonymous provider produces contributions nobody can revoke.

contribute is synchronous. Anything an extension needs to discover over the network happens before composition — the MCP adapter's discover() is its own async step — so composition itself is a pure function over values, replayable and comparable without an event loop.