Skip to content

symfonic.capabilities.knowledge.multimodal

multimodal

Multimodal normalisation and provider wire formatting — the outbound half.

Everything upstream of this module is about content coming in. This module is about the moment it goes back out, to a provider, and the T1.2.5 AS-NET contracts govern that moment.

No credential parameter (AS-NET-3) is enforced structurally: the wire formatters take exactly one argument — the parts. A function that cannot receive a credential cannot forward one, which is a stronger guarantee than remembering not to.

The media-type allowlists are the other half, and they are about the provider rather than the network: an attacker-chosen MIME string echoed onto the wire is an injection into the provider's own parser dispatch. There is one per media kind — images and documents — because a document's media type arrives on the same untrusted AttachmentRef an image's does.

AS-NET-1 and AS-NET-2 — the scheme allowlist and the host controls — live in :mod:.outbound, because they govern what the network stack may be asked for rather than what a provider may be told.

ContentPart dataclass 

ContentPart(part_type: PartType, text: str = '', media_type: str = '', data_b64: str = '')

One provider-neutral piece of a multimodal message.

media_part 

media_part(kind: str, media_type: str, payload: bytes) -> ContentPart

Normalise bytes into a base64 part, under the media-type allowlist.

Source code in src/symfonic/capabilities/knowledge/multimodal.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def media_part(kind: str, media_type: str, payload: bytes) -> ContentPart:
    """Normalise bytes into a base64 part, under the media-type allowlist."""
    if kind not in _MEDIA_KINDS:
        raise OutboundRejected(
            f"part kind {kind!r} has no wire representation; permitted kinds are "
            f"{sorted(_MEDIA_KINDS)}."
        )
    if not payload:
        raise OutboundRejected(f"the {kind} part carries an empty payload.")
    if not media_type:
        raise OutboundRejected(f"the {kind} part declares no media type.")
    allowed = _ALLOWED_MEDIA_TYPES[kind]
    if media_type not in allowed:
        raise OutboundRejected(
            f"{kind} media type {media_type!r} is outside the allowlist "
            f"{sorted(allowed)}. The media type travels to the provider and selects its "
            "decoder, so an attacker-chosen value is an injection into someone else's "
            "parser — and a document's media type comes from the same untrusted "
            "AttachmentRef an image's does."
        )
    return ContentPart(
        part_type="image" if kind == "image" else "document",
        media_type=media_type,
        data_b64=base64.b64encode(payload).decode("ascii"),
    )

text_part 

text_part(text: str) -> ContentPart

Normalise a text part. Whitespace-only text is refused, not sent.

Source code in src/symfonic/capabilities/knowledge/multimodal.py 
76
77
78
79
80
81
82
83
def text_part(text: str) -> ContentPart:
    """Normalise a text part. Whitespace-only text is refused, not sent."""
    stripped = text.strip()
    if not stripped:
        raise OutboundRejected(
            "a text part is empty; an empty block costs a request slot and says nothing."
        )
    return ContentPart(part_type="text", text=stripped)

to_anthropic_blocks 

to_anthropic_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]

Render parts as Anthropic content blocks.

Source code in src/symfonic/capabilities/knowledge/multimodal.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def to_anthropic_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]:
    """Render parts as Anthropic content blocks."""
    blocks: list[dict[str, object]] = []
    for part in parts:
        if part.part_type == "text":
            blocks.append({"type": "text", "text": part.text})
            continue
        blocks.append(
            {
                "type": part.part_type,
                "source": {
                    "type": "base64",
                    "media_type": part.media_type,
                    "data": part.data_b64,
                },
            }
        )
    return blocks

to_openai_blocks 

to_openai_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]

Render parts as OpenAI content blocks.

A document part is refused rather than approximated. OpenAI's chat content schema has no document block, and inventing one — a data: image URL carrying a PDF, say — produces a request the provider rejects at best and silently misreads at worst.

Source code in src/symfonic/capabilities/knowledge/multimodal.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def to_openai_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]:
    """Render parts as OpenAI content blocks.

    A document part is refused rather than approximated. OpenAI's chat content
    schema has no document block, and inventing one — a ``data:`` image URL
    carrying a PDF, say — produces a request the provider rejects at best and
    silently misreads at worst.
    """
    blocks: list[dict[str, object]] = []
    for part in parts:
        if part.part_type == "text":
            blocks.append({"type": "text", "text": part.text})
            continue
        if part.part_type == "document":
            raise OutboundRejected(
                "the OpenAI content schema has no document block; extract the document to "
                "text and send it as a text part rather than inventing a wire shape."
            )
        blocks.append(
            {
                "type": "image_url",
                "image_url": {"url": f"data:{part.media_type};base64,{part.data_b64}"},
            }
        )
    return blocks