Structured Tool Call Parser for
Autonomous AI Agents

from agent_tool_parser import try_parse_tool_calls

def run_agent_turn(messages: list[dict], client, model_name: str):
    response = client.chat.completions.create(
        model=model_name,
        messages=messages,
        temperature=0.0,
    )
    raw_content = response.choices[0].message.content or ""

    # Resilient extraction across DeepSeek, Claude, Qwen, Llama & non-standard JSON
    tool_calls = try_parse_tool_calls(
        raw_content,
        allowed_tools={"read_file", "write_file", "bash", "search"},
        tool_aliases={"view": "read_file", "sh": "bash", "cmd": "bash"}
    )

    if not tool_calls:
        return {"action": "reply", "content": raw_content}

    # Execute all tools (in parallel or sequentially)
    results = []
    for call in tool_calls:
        output = execute_tool(call.name, call.args)
        results.append({"tool": call.name, "output": output})

    return {"action": "continue", "results": results}

from agent_tool_parser import parse_tool_calls, safe_json_loads, ToolError

def normalize_response(response_message):
    """Normalizes response structures when inference endpoints emit markup in message content."""
    # 1. Return native tool_calls if already structured by provider SDK
    if getattr(response_message, "tool_calls", None):
        return [
            {"name": tc.function.name, "args": safe_json_loads(tc.function.arguments)}
            for tc in response_message.tool_calls
        ]

    # 2. Extract structured invocations from text content stream
    content = getattr(response_message, "content", "") or ""
    try:
        calls = parse_tool_calls(content)
        return [{"name": c.name, "args": c.args} for c in calls]
    except ToolError:
        return []

import asyncio
from agent_tool_parser import parse_tool_calls

async def handle_llm_turn(model_output: str):
    # Parse multiple parallel tool calls from one generation step
    calls = parse_tool_calls(model_output)

    # Execute all tools concurrently in parallel
    tasks = [async_run_tool(call.name, call.args) for call in calls]
    tool_results = await asyncio.gather(*tasks, return_exceptions=True)

    return list(zip([c.name for c in calls], tool_results))