symfonic.capabilities.extensions.mcp_reader

mcp_reader

Reading MCP wire payloads (T4.2.2).

The adapter next door decides what to do with an MCP server; this module decides what a payload says, and it is separate because those two failures have different fixes. An adapter bug is ours. A payload that fails to parse here belongs to whoever runs the server, and the message has to be good enough for them to act on without our source in front of them.

Two readings, one rule each.

:func:read_tool_payload refuses. A tool advertisement drives argument binding and routing, so a payload that is malformed in name or schema yields no tool at all — the caller records the refusal and moves to the next entry.

:func:read_result_text degrades. A result is content; a shape we do not understand is a result we did not get, and raising there would turn one confused server response into an exception inside the agent loop.

ToolDraft dataclass

ToolDraft(raw_name: str, description: str = '', input_schema: dict[str, Any] = dict(), truncated: bool = False)

One validated tool advertisement, before it is named and bound.

read_result_text

read_result_text(payload: Any, *, limits: TrustLimits) -> str

Read a tools/call result into bounded text.

Only type == "text" parts are read. The rest of the MCP content vocabulary (images, embedded resources) is dropped rather than stringified: a resource reference rendered into the transcript as its repr is a URL the model may act on, and integration output influences context, never targets (AS-INT-1).

A non-mapping payload, or one whose content is not a list, yields the empty string with no exception.

Both the part count and the running length are bounded as the parts are read, rather than only the finished join. Capping each part and joining an unbounded number of them still materialises their product first, which is the ceiling argument (AS-ING-6) applied one step too late. Whatever the bounds drop, the result carries :data:~.trust.TRUNCATION_MARKER, so a shortened answer never reads as a complete one.

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

def read_result_text(payload: Any, *, limits: TrustLimits) -> str:
    """Read a ``tools/call`` result into bounded text.

    Only ``type == "text"`` parts are read. The rest of the MCP content
    vocabulary (images, embedded resources) is *dropped rather than
    stringified*: a resource reference rendered into the transcript as its repr
    is a URL the model may act on, and integration output influences context,
    never targets (AS-INT-1).

    A non-mapping payload, or one whose ``content`` is not a list, yields the
    empty string with no exception.

    Both the part *count* and the running length are bounded as the parts are
    read, rather than only the finished join. Capping each part and joining an
    unbounded number of them still materialises their product first, which is
    the ceiling argument (AS-ING-6) applied one step too late. Whatever the
    bounds drop, the result carries :data:`~.trust.TRUNCATION_MARKER`, so a
    shortened answer never reads as a complete one.
    """
    if not isinstance(payload, Mapping):
        return ""
    parts = payload.get("content")
    if not isinstance(parts, Sequence) or isinstance(parts, (str, bytes)):
        return ""
    entries = list(islice(parts, limits.max_result_parts + 1))
    dropped = len(entries) > limits.max_result_parts
    del entries[limits.max_result_parts :]

    texts: list[str] = []
    remaining = limits.max_result_length
    for index, part in enumerate(entries):
        if remaining <= 0:
            dropped = dropped or any(_is_text_part(rest) for rest in entries[index:])
            break
        if not _is_text_part(part):
            continue
        text, truncated = sanitize_text(part.get("text"), limit=remaining)
        dropped = dropped or truncated
        if not text:
            continue
        # The newline the join will insert spends budget too.
        remaining -= len(text) + (1 if texts else 0)
        texts.append(text)

    joined = "\n".join(texts)
    if dropped and not joined.endswith(TRUNCATION_MARKER):
        joined += TRUNCATION_MARKER
    bounded, _ = sanitize_text(joined, limit=limits.max_result_length)
    return bounded

read_tool_payload

read_tool_payload(payload: Any, *, origin: str, limits: TrustLimits) -> ToolDraft

Validate one tools/list entry.

Raises:

Type	Description
UntrustedPayloadError	If the entry is not an object, its name is missing / mistyped / out of charset / over length, or its schema is not a bounded JSON object.

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

def read_tool_payload(
    payload: Any, *, origin: str, limits: TrustLimits
) -> ToolDraft:
    """Validate one ``tools/list`` entry.

    Raises:
        UntrustedPayloadError: If the entry is not an object, its name is
            missing / mistyped / out of charset / over length, or its schema is
            not a bounded JSON object.
    """
    if not isinstance(payload, Mapping):
        raise UntrustedPayloadError(
            f"{origin}: tool entry is {type(payload).__name__}, not an object."
        )
    raw_name = validate_identifier(
        payload.get("name"), field="tool name", origin=origin, limits=limits
    )
    schema = validate_schema(
        payload.get("inputSchema"),
        origin=origin,
        field=f"inputSchema of {raw_name!r}",
        limits=limits,
    )
    description, truncated = sanitize_text(
        payload.get("description"), limit=limits.max_description_length
    )
    return ToolDraft(
        raw_name=raw_name,
        description=description,
        input_schema=schema,
        truncated=truncated,
    )