Skip to content

symfonic.agent.response_messages

response_messages

Dumping kernel messages in the compatibility AgentResponse.messages shape.

AgentResponse.messages is a public payload: adopters index it, log it, and feed it back. The legacy engine fills it with LangChain message dumps ({"type": "ai", "content": ..., "tool_calls": [...]}) because the graph runtime hands it LangChain messages; the kernel hands its callers typed :class:~symfonic.agent.facade_types.Message values whose dump is {"role": ..., "tool_call_id": ...}. Each typed message is rebuilt as the LangChain message the legacy path would have produced, then dumped by the same serializer, so message["type"] keeps working on every route.

This native owner serves the compatibility HTTP routes directly and the cutover re-exports in :mod:symfonic.agent.cutover.messages and :mod:symfonic.agent.cutover.tool_names. It is the outbound direction only; the inbound history conversion stays with the cutover delegate.

Translation, not interpretation. :class:~symfonic.agent.facade_types.Message carries no message id and no response_metadata, so the rebuilt LangChain message gets LangChain's constructor defaults (None, {}) for both, where the legacy path gets whatever the provider put on the wire. Per-message usage_metadata survives (CORE-21). The two losses are pinned by tests/agent/cutover/test_facade_parity.py:: test_the_projection_does_not_carry_provider_message_metadata.

Tool messages are named the way the legacy body names them. Legacy's ToolMessage carries name=<tool>; the typed Message has no name field, so the name is recovered from the earlier assistant call with the same id. A tool message no earlier call names keeps None rather than a guess. A replayed tool message is reported exactly, from its tool_name (CORE-20).

as_legacy_messages 

as_legacy_messages(messages: Iterable[Any]) -> list[dict[str, Any]]

Dump kernel messages in the shape the legacy engine returns, tools named.

Source code in src/symfonic/agent/response_messages.py 
51
52
53
54
def as_legacy_messages(messages: Iterable[Any]) -> list[dict[str, Any]]:
    """Dump kernel messages in the shape the legacy engine returns, tools named."""
    typed = list(messages)
    return named_tool_messages(typed, [_as_langchain(message).model_dump() for message in typed])

named_tool_messages 

named_tool_messages(typed: Sequence[Any], dumped: list[dict[str, Any]]) -> list[dict[str, Any]]

Name each tool dump: replayed exactly, generated from an earlier call.

typed[i] is the message dumped[i] was built from.

Source code in src/symfonic/agent/response_messages.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def named_tool_messages(typed: Sequence[Any], dumped: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Name each ``tool`` dump: replayed exactly, generated from an earlier call.

    ``typed[i]`` is the message ``dumped[i]`` was built from.
    """
    names: dict[str, str] = {}
    for message, dump in zip(typed, dumped, strict=True):
        if dump.get("type") == "ai":
            names.update(names_by_call_id(dump.get("tool_calls") or ()))
        elif dump.get("type") == "tool":
            dump["name"] = (
                message.tool_name
                if hasattr(message, "tool_name")
                else names.get(dump.get("tool_call_id") or "")
            )
    return dumped