Parsers

Built-in response parsers for normalizing provider-specific tool call formats.

parsers

Built-in response parsers for normalizing provider-specific tool call formats.

GeminiResponseParser

Parses Gemini GenerateContentResponse into AgentResponse.

Handles the Gemini response format where tool calls appear as function_call parts in response.candidates[].content.parts[].

Works with both the raw dict format and the google-genai SDK objects.

Usage

parser = GeminiResponseParser() response = parser.parse(gemini_raw_response)

parse

parse(raw_response: Any) -> AgentResponse

Parse a Gemini response into a normalized AgentResponse.

Source code in src/russo/parsers/gemini.py

def parse(self, raw_response: Any) -> AgentResponse:
    """Parse a Gemini response into a normalized AgentResponse."""
    tool_calls: list[ToolCall] = []

    # Handle google-genai SDK response objects
    candidates = _get_attr_or_key(raw_response, "candidates", [])
    for candidate in candidates:
        content = _get_attr_or_key(candidate, "content", None)
        if content is None:
            continue
        parts = _get_attr_or_key(content, "parts", [])
        for part in parts:
            fc = _get_attr_or_key(part, "function_call", None)
            if fc is not None:
                name = _get_attr_or_key(fc, "name", "")
                args = _get_attr_or_key(fc, "args", {})
                if isinstance(args, str):
                    import json

                    args = json.loads(args)
                tool_calls.append(ToolCall(name=name, arguments=dict(args) if args else {}))

    return AgentResponse(tool_calls=tool_calls, raw=raw_response)

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

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

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)

OpenAIResponseParser

Parses OpenAI ChatCompletion responses into AgentResponse.

Handles the OpenAI format where tool calls appear at: response.choices[].message.tool_calls[]

Each tool call has: {id, type, function: {name, arguments}}.

Works with both the raw dict format and the openai SDK objects.

Usage

parser = OpenAIResponseParser() response = parser.parse(openai_raw_response)

parse

parse(raw_response: Any) -> AgentResponse

Parse an OpenAI response into a normalized AgentResponse.

Source code in src/russo/parsers/openai.py

def parse(self, raw_response: Any) -> AgentResponse:
    """Parse an OpenAI response into a normalized AgentResponse."""
    tool_calls: list[ToolCall] = []

    choices = _get_attr_or_key(raw_response, "choices", [])
    for choice in choices:
        message = _get_attr_or_key(choice, "message", None)
        if message is None:
            continue
        raw_tool_calls = _get_attr_or_key(message, "tool_calls", [])
        if not raw_tool_calls:
            continue
        for tc in raw_tool_calls:
            function = _get_attr_or_key(tc, "function", None)
            if function is None:
                continue
            name = _get_attr_or_key(function, "name", "")
            arguments_raw = _get_attr_or_key(function, "arguments", "{}")
            if isinstance(arguments_raw, str):
                try:
                    arguments = json.loads(arguments_raw)
                except (json.JSONDecodeError, TypeError):
                    arguments = {}
            elif isinstance(arguments_raw, dict):
                arguments = arguments_raw
            else:
                arguments = {}
            tool_calls.append(ToolCall(name=name, arguments=arguments))

    return AgentResponse(tool_calls=tool_calls, raw=raw_response)