Skip to content

JSON Response Parser

Configurable parser for custom HTTP and WebSocket response structures.

mapping

Configurable parser for custom JSON response structures.

JsonResponseParser 

JsonResponseParser(*, tool_calls_key: str = 'tool_calls', name_key: str = 'name', arguments_key: str = 'arguments', single: bool = False)

Configurable parser for endpoints that return tool calls in a custom JSON structure.

Instead of writing a full parser class, pass field name config to match whatever your HTTP or WebSocket endpoint returns.

PARAMETER DESCRIPTION
tool_calls_key

Dot-separated path to the tool calls in the response. Supports nested paths (e.g. "result.toolCalls"). Default: "tool_calls".

TYPE: str DEFAULT: 'tool_calls'

name_key

Key for the function name within each tool call dict. Default: "name".

TYPE: str DEFAULT: 'name'

arguments_key

Key for the arguments dict within each tool call. Default: "arguments".

TYPE: str DEFAULT: 'arguments'

single

Set to True if the endpoint returns a single tool call object instead of a list. Default: False.

TYPE: bool DEFAULT: False

Usage::

# Endpoint returns: {"toolCall": {"name": "fn", "arguments": {...}}}
parser = JsonResponseParser(tool_calls_key="toolCall", single=True)

# Endpoint returns: {"result": {"calls": [{"fn": "...", "params": {...}}]}}
parser = JsonResponseParser(
    tool_calls_key="result.calls",
    name_key="fn",
    arguments_key="params",
)

# Use with HttpAgent or WebSocketAgent
agent = HttpAgent(url="http://localhost:8000/agent", parser=parser)
agent = WebSocketAgent(url="ws://localhost:8000/ws", parser=parser)
Source code in src/russo/parsers/mapping.py 
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    *,
    tool_calls_key: str = "tool_calls",
    name_key: str = "name",
    arguments_key: str = "arguments",
    single: bool = False,
) -> None:
    self.tool_calls_key = tool_calls_key
    self.name_key = name_key
    self.arguments_key = arguments_key
    self.single = single

parse 

parse(raw_response: Any) -> AgentResponse

Parse a JSON response into a normalized AgentResponse.

Source code in src/russo/parsers/mapping.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
def parse(self, raw_response: Any) -> AgentResponse:
    """Parse a JSON response into a normalized AgentResponse."""
    # If the response is a list (e.g. aggregated WebSocket messages),
    # search each item and return on first hit.
    if isinstance(raw_response, list):
        for item in raw_response:
            result = self._try_parse(item)
            if result is not None:
                return AgentResponse(tool_calls=result, raw=raw_response)
        return AgentResponse(tool_calls=[], raw=raw_response)

    tool_calls = self._try_parse(raw_response)
    return AgentResponse(tool_calls=tool_calls or [], raw=raw_response)