technique_router

technique_router ¶

Technique router — classifies agent output and routes to handlers.

After a musician executes a sheet through a backend, the technique router inspects the output to determine what kind of result was produced and routes it to the appropriate handler:

prose — standard text output, no special routing needed
code_block — executable code that should run in a sandbox
tool_call — MCP tool invocation, route to shared MCP pool
a2a_request — inter-agent task delegation, route through event bus

For MCP-native instruments (claude-code, gemini-cli), tool calls go through the instrument's native MCP support. The technique router handles bridging for non-MCP-native instruments (OpenRouter free models) that produce code or structured output.

Classification uses pattern matching on the output text — no LLM calls. The patterns are conservative: when in doubt, classify as prose (the safe default that doesn't trigger any special routing).

See: design spec sections 8.2 (Code Mode) and 8.3 (Technique Router)

Classes¶

OutputKind ¶

Bases: str, Enum

Classification of agent output for routing decisions.

Attributes¶

PROSE `class-attribute` `instance-attribute` ¶

PROSE = 'prose'

Standard text output — no special routing.

CODE_BLOCK `class-attribute` `instance-attribute` ¶

CODE_BLOCK = 'code_block'

Executable code in markdown fences — route to sandbox.

TOOL_CALL `class-attribute` `instance-attribute` ¶

TOOL_CALL = 'tool_call'

MCP tool invocation — route to shared MCP pool.

A2A_REQUEST `class-attribute` `instance-attribute` ¶

A2A_REQUEST = 'a2a_request'

Inter-agent task delegation — route through event bus.

ClassifiedOutput `dataclass` ¶

ClassifiedOutput(kind, raw_output, code_blocks=list(), tool_calls=list(), a2a_requests=list())

Result of classifying agent output.

Contains the classification and extracted content for routing. For code blocks, code_blocks contains the extracted code. For tool calls, tool_calls contains the parsed invocations. For A2A, a2a_requests contains parsed delegation requests.

CodeBlock `dataclass` ¶

CodeBlock(language, code)

An extracted code block from agent output.

Attributes:

Name	Type	Description
`language`	`str`	The language tag from the code fence (e.g., "python").
`code`	`str`	The code content between the fences.

ToolCallRequest `dataclass` ¶

ToolCallRequest(server, method, arguments=dict())

A parsed MCP tool call from agent output.

Attributes:

Name	Type	Description
`server`	`str`	The MCP server name (e.g., "github", "filesystem").
`method`	`str`	The method to invoke (e.g., "list_issues", "read_file").
`arguments`	`dict[str, str]`	Parsed arguments as key-value pairs.

A2ARoutingRequest `dataclass` ¶

A2ARoutingRequest(target_agent, task_description, context=dict())

A parsed A2A task delegation from agent output.

Attributes:

Name	Type	Description
`target_agent`	`str`	The agent to delegate to.
`task_description`	`str`	What needs to be done.
`context`	`dict[str, str]`	Additional context for the task.

TechniqueRouter ¶

Classifies agent output and determines routing.

The router inspects output text and applies pattern matching to determine whether the output contains executable code, tool calls, A2A requests, or is plain prose.

Classification priority (first match wins): 1. A2A requests — checked first because they're explicit directives 2. Tool calls — explicit tool invocations 3. Code blocks — executable code in fences 4. Prose — default fallback

Usage::

router = TechniqueRouter()
result = router.classify(agent_output)

match result.kind:
    case OutputKind.CODE_BLOCK:
        for block in result.code_blocks:
            sandbox.execute(block.code)
    case OutputKind.TOOL_CALL:
        for call in result.tool_calls:
            mcp_pool.invoke(call.server, call.method, call.arguments)
    case OutputKind.A2A_REQUEST:
        for req in result.a2a_requests:
            event_bus.submit_a2a_task(req)
    case OutputKind.PROSE:
        pass  # standard output

Functions¶

classify ¶

classify(output)

Classify agent output for routing.

Parameters:

Name	Type	Description	Default
`output`	`str`	Raw text output from the musician/backend.	required

Returns:

Type	Description
`ClassifiedOutput`	ClassifiedOutput with kind and extracted content.

Source code in src/marianne/daemon/technique_router.py

def classify(self, output: str) -> ClassifiedOutput:
    """Classify agent output for routing.

    Args:
        output: Raw text output from the musician/backend.

    Returns:
        ClassifiedOutput with kind and extracted content.
    """
    if not output or not output.strip():
        return ClassifiedOutput(kind=OutputKind.PROSE, raw_output=output or "")

    # Check for A2A delegation requests
    a2a_requests = self._extract_a2a_requests(output)
    if a2a_requests:
        _logger.debug(
            "technique_router.classified",
            extra={"kind": "a2a_request", "count": len(a2a_requests)},
        )
        return ClassifiedOutput(
            kind=OutputKind.A2A_REQUEST,
            raw_output=output,
            a2a_requests=a2a_requests,
        )

    # Check for tool calls
    tool_calls = self._extract_tool_calls(output)
    if tool_calls:
        _logger.debug(
            "technique_router.classified",
            extra={"kind": "tool_call", "count": len(tool_calls)},
        )
        return ClassifiedOutput(
            kind=OutputKind.TOOL_CALL,
            raw_output=output,
            tool_calls=tool_calls,
        )

    # Check for executable code blocks
    code_blocks = self._extract_code_blocks(output)
    if code_blocks:
        _logger.debug(
            "technique_router.classified",
            extra={"kind": "code_block", "count": len(code_blocks)},
        )
        return ClassifiedOutput(
            kind=OutputKind.CODE_BLOCK,
            raw_output=output,
            code_blocks=code_blocks,
        )

    # Default: prose
    return ClassifiedOutput(kind=OutputKind.PROSE, raw_output=output)

technique_router