symfonic.capabilities.extensions.ports

ports

The ports extension composition binds (T4.2.2).

Two protocols, both narrow, both structural.

:class:McpServerPort is what the adapter needs from an MCP server and nothing else: list, call, close. It speaks raw mappings, deliberately — the wire answers JSON, and a port typed in terms of already-parsed framework objects would push parsing into the transport, which is the layer least able to refuse a bad payload. Parsing and refusal belong at the boundary (:mod:~symfonic.capabilities.extensions.trust), so the port hands the boundary what actually arrived.

:class:LegacyPluginPort is what the compatibility bridge reads from a plugin written against the pre-refactor API. It is listed here rather than imported from symfonic.core.plugins for two reasons: the capability layer does not import legacy packages, and a structural declaration is the honest one — the bridge works on anything shaped like a plugin, including the many adopter plugins that never imported the framework's Protocol at all.

Neither protocol is runtime_checkable in the "every member present" sense the framework usually wants: LegacyPluginPort describes a surface where every member is optional in practice (a plugin may define inject_contributions or inject_system_prompt), so the bridge probes for members rather than asserting the whole shape. Saying so here is cheaper than a reader discovering it from a missing isinstance.

LegacyPluginPort

Bases: Protocol

The pre-refactor domain-plugin surface, declared structurally.

Every member is optional at runtime; the bridge probes. The declaration exists so the expected signatures are written down somewhere a plugin author can read, and so a type checker can flag a plugin whose validate_state_transition takes the wrong arity before it silently fails open on every turn.

McpServerPort

Bases: Protocol

One MCP server, as the adapter uses it.

An implementation answers raw JSON-shaped mappings; every field is untrusted until the boundary has read it.

call_tool async

call_tool(tool_name: str, arguments: Mapping[str, Any]) -> Mapping[str, Any]

Invoke one tool and return the raw result payload.

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

async def call_tool(
    self, tool_name: str, arguments: Mapping[str, Any]
) -> Mapping[str, Any]:
    """Invoke one tool and return the raw result payload."""
    ...

close async

close() -> None

Release the connection. Called by the composition's teardown hook.

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

async def close(self) -> None:
    """Release the connection. Called by the composition's teardown hook."""
    ...

list_tools async

list_tools() -> Sequence[Mapping[str, Any]]

Return the server's advertised tools as raw payloads.

Each payload is expected to carry name and optionally description / inputSchema, but the adapter assumes nothing: a payload that is not a mapping is refused, not coerced.

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

async def list_tools(self) -> Sequence[Mapping[str, Any]]:
    """Return the server's advertised tools as raw payloads.

    Each payload is expected to carry ``name`` and optionally
    ``description`` / ``inputSchema``, but the adapter assumes nothing:
    a payload that is not a mapping is refused, not coerced.
    """
    ...