Skip to content

coordinator.parser

coordinator.parser

Structured output parser for coordinator agent responses.

Agents produce text with recognized KEY: VALUE blocks. This module extracts them into a normalized dict for programmatic access.

parse_agent_output 

parse_agent_output(raw: str, agent: str = '') -> dict[str, str]

Extract structured KEY: VALUE blocks from agent output text.

Parameters:

Name Type Description Default
raw str

The full text output from a subagent.

 required
agent str

Agent type (unused currently, reserved for future agent-specific parsing rules).

 ''

Returns:

Type Description
dict[str, str]

Dict mapping lowercase key names to their stripped string values.
dict[str, str]

If a key appears multiple times, the last occurrence wins.
dict[str, str]

Returns an empty dict if raw is empty or None.
Source code in coordinator/parser.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def parse_agent_output(raw: str, agent: str = "") -> dict[str, str]:
    """Extract structured KEY: VALUE blocks from agent output text.

    Args:
        raw: The full text output from a subagent.
        agent: Agent type (unused currently, reserved for future
               agent-specific parsing rules).

    Returns:
        Dict mapping lowercase key names to their stripped string values.
        If a key appears multiple times, the last occurrence wins.
        Returns an empty dict if raw is empty or None.
    """
    if not raw:
        return {}

    raw = raw.replace("\r\n", "\n").replace("\r", "\n")

    result: dict[str, str] = {}
    lines = raw.split("\n")

    current_key: str | None = None
    current_lines: list[str] = []

    for line in lines:
        match = _KEY_PATTERN.match(line)
        if match:
            key = match.group(1)
            if (
                key.lower() in result
                and current_key is not None
                and current_key in MULTI_LINE_KEYS
            ):
                current_lines.append(line)
                continue
            if current_key is not None:
                result[current_key.lower()] = _join_value(current_key, current_lines)
            current_key = key
            first_value = match.group(2)
            current_lines = [first_value] if first_value else []
        elif current_key is not None and current_key in MULTI_LINE_KEYS:
            current_lines.append(line)

    if current_key is not None:
        result[current_key.lower()] = _join_value(current_key, current_lines)

    return result