Add native Responses API transport

Route Agent Zero turns through a LiteLLM transport layer that prefers the Responses API while preserving chat-completions fallback for providers without compatible endpoints.

Persist Responses metadata in history and agent state so provider-state continuation, local replay, native function-call execution, and stored-response cleanup survive normal chat workflows.

Normalize prompt caching by provider: OpenAI and Azure use prompt_cache_key and prompt_cache_retention, while Anthropic, Gemini, Bedrock, OpenRouter, and compatible chat providers keep block-level cache_control breakpoints and cached tool definitions.

helpers/history.py

@@ -40,9 +40,12 @@ MessageContent = Union[
 ]
 
 
-class OutputMessage(TypedDict):
+class OutputMessage(TypedDict, total=False):
     ai: bool
     content: MessageContent
+    metadata: dict[str, Any]
+    id: str
+    sequence: int
 
 
 class Record:
@@ -82,10 +85,20 @@ class Record:
 
 
 class Message(Record):
-    def __init__(self, ai: bool, content: MessageContent, tokens: int = 0, id: str = ""):
+    def __init__(
+        self,
+        ai: bool,
+        content: MessageContent,
+        tokens: int = 0,
+        id: str = "",
+        metadata: dict[str, Any] | None = None,
+        sequence: int = 0,
+    ):
         self.id = id or str(uuid.uuid4())
         self.ai = ai
         self.content = content
+        self.metadata = metadata or {}
+        self.sequence = sequence
         self.summary: str = ""
         self.tokens: int = tokens or self.calculate_tokens()
 
@@ -106,7 +119,15 @@ class Message(Record):
         return False
 
     def output(self):
-        return [OutputMessage(ai=self.ai, content=self.summary or self.content)]
+        return [
+            OutputMessage(
+                ai=self.ai,
+                content=self.summary or self.content,
+                metadata=self.metadata,
+                id=self.id,
+                sequence=self.sequence,
+            )
+        ]
 
     def output_langchain(self):
         return output_langchain(self.output())
@@ -120,6 +141,8 @@ class Message(Record):
             "id": self.id,
             "ai": self.ai,
             "content": self.content,
+            "metadata": self.metadata,
+            "sequence": self.sequence,
             "summary": self.summary,
             "tokens": self.tokens,
         }
@@ -127,7 +150,13 @@ class Message(Record):
     @staticmethod
     def from_dict(data: dict, history: "History"):
         content = data.get("content", "Content lost")
-        msg = Message(ai=data["ai"], content=content, id=data.get("id", ""))
+        msg = Message(
+            ai=data["ai"],
+            content=content,
+            id=data.get("id", ""),
+            metadata=data.get("metadata", {}) if isinstance(data.get("metadata"), dict) else {},
+            sequence=int(data.get("sequence", 0) or 0),
+        )
         msg.summary = data.get("summary", "")
         msg.tokens = data.get("tokens", 0)
         return msg
@@ -146,9 +175,22 @@ class Topic(Record):
             return sum(msg.get_tokens() for msg in self.messages)
 
     def add_message(
-        self, ai: bool, content: MessageContent, tokens: int = 0, id: str = ""
+        self,
+        ai: bool,
+        content: MessageContent,
+        tokens: int = 0,
+        id: str = "",
+        metadata: dict[str, Any] | None = None,
+        sequence: int = 0,
     ) -> Message:
-        msg = Message(ai=ai, content=content, tokens=tokens, id=id)
+        msg = Message(
+            ai=ai,
+            content=content,
+            tokens=tokens,
+            id=id,
+            metadata=metadata,
+            sequence=sequence,
+        )
         self.messages.append(msg)
         return msg
 
@@ -335,10 +377,22 @@ class History(Record):
         return self.current.get_tokens()
 
     def add_message(
-        self, ai: bool, content: MessageContent, tokens: int = 0, id: str = ""
+        self,
+        ai: bool,
+        content: MessageContent,
+        tokens: int = 0,
+        id: str = "",
+        metadata: dict[str, Any] | None = None,
     ) -> Message:
         self.counter += 1
-        return self.current.add_message(ai, content=content, tokens=tokens, id=id)
+        return self.current.add_message(
+            ai,
+            content=content,
+            tokens=tokens,
+            id=id,
+            metadata=metadata,
+            sequence=self.counter,
+        )
 
     def new_topic(self):
         if self.current.messages:
@@ -353,6 +407,35 @@ class History(Record):
         result += self.current.output()
         return result
 
+    def messages_since(self, sequence: int) -> list[Message]:
+        return [
+            message
+            for message in self.all_messages()
+            if int(message.sequence or 0) > int(sequence or 0)
+        ]
+
+    def all_messages(self) -> list[Message]:
+        messages: list[Message] = []
+        for bulk in self.bulks:
+            messages.extend(_messages_from_record(bulk))
+        for topic in self.topics:
+            messages.extend(topic.messages)
+        messages.extend(self.current.messages)
+        return messages
+
+    def latest_llm_result_for_model(self, provider_model_key: str):
+        from helpers.llm_result import result_from_metadata
+
+        for message in reversed(self.all_messages()):
+            if not message.ai:
+                continue
+            result = result_from_metadata(message.metadata)
+            if not result:
+                continue
+            if result.provider_model_key == provider_model_key and result.response_id:
+                return result
+        return None
+
     def trim_embeds(self, max_embeds: int) -> int:
         if max_embeds == -1:
             return 0
@@ -679,6 +762,19 @@ def _is_embedded_data(obj: object) -> bool:
     return isinstance(obj, Mapping) and obj.get("type") == "image_url"
 
 
+def _messages_from_record(record: Record) -> list[Message]:
+    if isinstance(record, Message):
+        return [record]
+    if isinstance(record, Topic):
+        return list(record.messages)
+    if isinstance(record, Bulk):
+        messages: list[Message] = []
+        for nested in record.records:
+            messages.extend(_messages_from_record(nested))
+        return messages
+    return []
+
+
 def _json_dumps(obj):
     return json.dumps(obj, ensure_ascii=False)
 

helpers/llm_result.py

@@ -0,0 +1,311 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import json
+from typing import Any
+
+
+RESPONSE_METADATA_KEY = "responses"
+LOCAL_FUNCTION_TOOL_TYPES = {"function_call"}
+TEXT_OUTPUT_TYPES = {"message"}
+REASONING_OUTPUT_TYPES = {"reasoning"}
+
+
+@dataclass
+class ResponseItem:
+    type: str
+    data: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_any(cls, item: Any) -> "ResponseItem":
+        data = object_to_dict(item)
+        return cls(type=str(data.get("type") or ""), data=data)
+
+    def to_dict(self) -> dict[str, Any]:
+        return dict(self.data)
+
+
+@dataclass
+class ResponseFunctionCall:
+    name: str
+    arguments: dict[str, Any]
+    call_id: str
+    item_id: str = ""
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_item(cls, item: ResponseItem) -> "ResponseFunctionCall | None":
+        if item.type != "function_call":
+            return None
+        name = str(item.data.get("name") or "")
+        if not name:
+            return None
+        return cls(
+            name=name,
+            arguments=parse_arguments(item.data.get("arguments")),
+            call_id=str(item.data.get("call_id") or item.data.get("id") or ""),
+            item_id=str(item.data.get("id") or ""),
+            raw=dict(item.data),
+        )
+
+
+@dataclass
+class LLMResult:
+    response: str = ""
+    reasoning: str = ""
+    response_id: str = ""
+    previous_response_id: str = ""
+    input_items: list[dict[str, Any]] = field(default_factory=list)
+    output_items: list[ResponseItem] = field(default_factory=list)
+    provider_model_key: str = ""
+    mode: str = "responses"
+    state: str = "provider"
+    usage: dict[str, Any] = field(default_factory=dict)
+    raw: dict[str, Any] = field(default_factory=dict)
+    capability: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult":
+        data = data or {}
+        return cls(
+            response=str(data.get("response") or ""),
+            reasoning=str(data.get("reasoning") or ""),
+            response_id=str(data.get("response_id") or ""),
+            previous_response_id=str(data.get("previous_response_id") or ""),
+            input_items=list(data.get("input_items") or []),
+            output_items=[
+                ResponseItem.from_any(item) for item in data.get("output_items") or []
+            ],
+            provider_model_key=str(data.get("provider_model_key") or ""),
+            mode=str(data.get("mode") or "responses"),
+            state=str(data.get("state") or "provider"),
+            usage=object_to_dict(data.get("usage") or {}),
+            raw=object_to_dict(data.get("raw") or {}),
+            capability=object_to_dict(data.get("capability") or {}),
+        )
+
+    @classmethod
+    def from_response(
+        cls,
+        response: Any,
+        *,
+        input_items: list[dict[str, Any]] | None = None,
+        previous_response_id: str = "",
+        provider_model_key: str = "",
+        mode: str = "responses",
+        state: str = "provider",
+        capability: dict[str, Any] | None = None,
+    ) -> "LLMResult":
+        raw = object_to_dict(response)
+        output_items = [ResponseItem.from_any(item) for item in as_list(raw.get("output"))]
+        result = cls(
+            response_id=str(raw.get("id") or ""),
+            previous_response_id=str(
+                raw.get("previous_response_id") or previous_response_id or ""
+            ),
+            input_items=list(input_items or []),
+            output_items=output_items,
+            provider_model_key=provider_model_key,
+            mode=mode,
+            state=state,
+            usage=object_to_dict(raw.get("usage") or {}),
+            raw=raw,
+            capability=dict(capability or {}),
+        )
+        result.response = output_text(raw, output_items)
+        result.reasoning = reasoning_text(output_items)
+        if not result.response and result.function_calls:
+            result.response = result.function_calls_text()
+        return result
+
+    @classmethod
+    def from_chat(
+        cls,
+        *,
+        response: str,
+        reasoning: str = "",
+        input_items: list[dict[str, Any]] | None = None,
+        provider_model_key: str = "",
+        capability: dict[str, Any] | None = None,
+    ) -> "LLMResult":
+        output_items = []
+        if response:
+            output_items.append(
+                ResponseItem(
+                    type="message",
+                    data={
+                        "type": "message",
+                        "role": "assistant",
+                        "content": [{"type": "output_text", "text": response}],
+                    },
+                )
+            )
+        if reasoning:
+            output_items.insert(
+                0,
+                ResponseItem(
+                    type="reasoning",
+                    data={
+                        "type": "reasoning",
+                        "summary": [{"type": "summary_text", "text": reasoning}],
+                    },
+                ),
+            )
+        return cls(
+            response=response,
+            reasoning=reasoning,
+            input_items=list(input_items or []),
+            output_items=output_items,
+            provider_model_key=provider_model_key,
+            mode="chat_completions",
+            state="off",
+            capability=dict(capability or {}),
+        )
+
+    @property
+    def function_calls(self) -> list[ResponseFunctionCall]:
+        calls: list[ResponseFunctionCall] = []
+        for item in self.output_items:
+            call = ResponseFunctionCall.from_item(item)
+            if call:
+                calls.append(call)
+        return calls
+
+    @property
+    def builtin_items(self) -> list[ResponseItem]:
+        return [
+            item
+            for item in self.output_items
+            if item.type
+            and item.type not in TEXT_OUTPUT_TYPES
+            and item.type not in REASONING_OUTPUT_TYPES
+            and item.type not in LOCAL_FUNCTION_TOOL_TYPES
+        ]
+
+    def function_calls_text(self) -> str:
+        calls = [
+            {"tool_name": call.name, "tool_args": call.arguments}
+            for call in self.function_calls
+        ]
+        if not calls:
+            return ""
+        if len(calls) == 1:
+            return json.dumps(calls[0])
+        return json.dumps(
+            {"tool_name": "parallel_tool_calls", "tool_args": {"calls": calls}}
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "response": self.response,
+            "reasoning": self.reasoning,
+            "response_id": self.response_id,
+            "previous_response_id": self.previous_response_id,
+            "input_items": self.input_items,
+            "output_items": [item.to_dict() for item in self.output_items],
+            "provider_model_key": self.provider_model_key,
+            "mode": self.mode,
+            "state": self.state,
+            "usage": self.usage,
+            "raw": self.raw,
+            "capability": self.capability,
+        }
+
+    def metadata(self) -> dict[str, Any]:
+        return {RESPONSE_METADATA_KEY: self.to_dict()}
+
+
+def function_call_output_item(
+    call_id: str,
+    output: str,
+    *,
+    acknowledged_safety_checks: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    item: dict[str, Any] = {
+        "type": "function_call_output",
+        "call_id": str(call_id or ""),
+        "output": output,
+    }
+    if acknowledged_safety_checks:
+        item["acknowledged_safety_checks"] = acknowledged_safety_checks
+    return item
+
+
+def metadata_from_llm_result(result: LLMResult | None) -> dict[str, Any]:
+    return result.metadata() if result else {}
+
+
+def result_from_metadata(metadata: dict[str, Any] | None) -> LLMResult | None:
+    if not isinstance(metadata, dict):
+        return None
+    data = metadata.get(RESPONSE_METADATA_KEY)
+    if not isinstance(data, dict):
+        return None
+    return LLMResult.from_dict(data)
+
+
+def object_to_dict(obj: Any) -> dict[str, Any]:
+    if isinstance(obj, dict):
+        return dict(obj)
+    if hasattr(obj, "model_dump"):
+        dumped = obj.model_dump()
+        return dict(dumped) if isinstance(dumped, dict) else {}
+    if hasattr(obj, "dict"):
+        dumped = obj.dict()
+        return dict(dumped) if isinstance(dumped, dict) else {}
+    return {}
+
+
+def as_list(value: Any) -> list[Any]:
+    return value if isinstance(value, list) else []
+
+
+def output_text(raw: dict[str, Any], output_items: list[ResponseItem]) -> str:
+    direct = raw.get("output_text")
+    if isinstance(direct, str):
+        return direct
+    pieces: list[str] = []
+    for item in output_items:
+        if item.type != "message":
+            continue
+        for block in as_list(item.data.get("content")):
+            if not isinstance(block, dict):
+                continue
+            block_type = block.get("type")
+            if block_type in {"output_text", "text", "input_text"}:
+                text = block.get("text")
+                if isinstance(text, str):
+                    pieces.append(text)
+            elif block_type == "refusal":
+                refusal = block.get("refusal")
+                if isinstance(refusal, str):
+                    pieces.append(refusal)
+    return "".join(pieces)
+
+
+def reasoning_text(output_items: list[ResponseItem]) -> str:
+    pieces: list[str] = []
+    for item in output_items:
+        if item.type != "reasoning":
+            continue
+        for block in as_list(item.data.get("summary")):
+            if isinstance(block, dict):
+                text = block.get("text") or block.get("reasoning")
+                if isinstance(text, str):
+                    pieces.append(text)
+            elif isinstance(block, str):
+                pieces.append(block)
+    return "".join(pieces)
+
+
+def parse_arguments(raw_arguments: Any) -> dict[str, Any]:
+    if isinstance(raw_arguments, dict):
+        return raw_arguments
+    if isinstance(raw_arguments, str):
+        try:
+            parsed = json.loads(raw_arguments or "{}")
+        except Exception:
+            parsed = {"arguments": raw_arguments}
+    else:
+        parsed = {"arguments": raw_arguments}
+    return parsed if isinstance(parsed, dict) else {"arguments": parsed}

helpers/persist_chat.py

@@ -4,6 +4,7 @@ from typing import Any
 import uuid
 from agent import Agent, AgentConfig, AgentContext, AgentContextType
 from helpers import files, history
+from helpers.litellm_transport import delete_stored_response_ids
 from helpers.localization import Localization
 import json
 from initialize import initialize_agent
@@ -118,6 +119,7 @@ def export_json_chat(context: AgentContext):
 
 def remove_chat(ctxid):
     """Remove a chat or task context"""
+    _delete_provider_responses_for_chat(ctxid)
     path = get_chat_folder_path(ctxid)
     files.delete_dir(path)
 
@@ -324,3 +326,70 @@ def _safe_json_serialize(obj, **kwargs):
             return False
 
     return json.dumps(obj, default=serializer, **kwargs)
+
+
+def _delete_provider_responses_for_chat(ctxid: str) -> None:
+    try:
+        data = json.loads(files.read_file(_get_chat_file_path(ctxid)))
+    except Exception:
+        return
+    if _responses_delete_disabled(data):
+        return
+    response_ids = _collect_response_ids(data)
+    if not response_ids:
+        return
+    delete_stored_response_ids(response_ids)
+
+
+def _responses_delete_disabled(data: dict[str, Any]) -> bool:
+    if data.get("responses_delete_on_chat_delete") is False:
+        return True
+    context_data = data.get("data")
+    if isinstance(context_data, dict) and context_data.get("responses_delete_on_chat_delete") is False:
+        return True
+    for agent_data in data.get("agents", []) or []:
+        if not isinstance(agent_data, dict):
+            continue
+        state = agent_data.get("data")
+        if isinstance(state, dict) and state.get("responses_delete_on_chat_delete") is False:
+            return True
+    return False
+
+
+def _collect_response_ids(data: Any) -> list[str]:
+    found: list[str] = []
+    seen: set[str] = set()
+
+    def add(value: Any) -> None:
+        response_id = str(value or "").strip()
+        if response_id and response_id not in seen:
+            seen.add(response_id)
+            found.append(response_id)
+
+    def walk(obj: Any) -> None:
+        if isinstance(obj, dict):
+            state = obj.get(Agent.DATA_NAME_RESPONSES_STATE)
+            if isinstance(state, dict):
+                add(state.get("response_id"))
+                for response_id in state.get("response_ids") or []:
+                    add(response_id)
+
+            metadata = obj.get("metadata")
+            if isinstance(metadata, dict):
+                responses = metadata.get("responses")
+                if isinstance(responses, dict):
+                    add(responses.get("response_id"))
+
+            for value in obj.values():
+                walk(value)
+        elif isinstance(obj, list):
+            for value in obj:
+                walk(value)
+        elif isinstance(obj, str) and '"response_id"' in obj:
+            try:
+                walk(json.loads(obj))
+            except Exception:
+                return
+
+    walk(data)
+    return found

helpers/responses_tools.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+from typing import Any
+
+from helpers import files, subagents
+
+
+FUNCTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
+TOOL_PROMPT_PREFIX = "agent.system.tool."
+TOOL_PROMPT_SUFFIX = ".md"
+MAX_TOOL_DESCRIPTION_CHARS = 1024
+
+
+def build_responses_function_tools(agent: Any) -> tuple[list[dict[str, Any]], dict[str, str]]:
+    """Build permissive Responses function tools from A0 tool prompts and MCP schemas."""
+
+    tools: list[dict[str, Any]] = []
+    name_map: dict[str, str] = {}
+
+    for tool_name, prompt in _local_tool_prompts(agent):
+        native_name = _native_tool_name(tool_name)
+        name_map[native_name] = tool_name
+        tools.append(
+            {
+                "type": "function",
+                "name": native_name,
+                "description": _description_from_prompt(prompt, fallback=tool_name),
+                "parameters": _schema_from_prompt(prompt),
+            }
+        )
+
+    for tool_name, tool in _mcp_tools(agent):
+        native_name = _native_tool_name(tool_name)
+        name_map[native_name] = tool_name
+        tools.append(
+            {
+                "type": "function",
+                "name": native_name,
+                "description": _truncate(str(tool.get("description") or tool_name)),
+                "parameters": _schema_from_any(tool.get("input_schema")),
+            }
+        )
+
+    return _dedupe_tools(tools), name_map
+
+
+def original_tool_name(native_name: str, name_map: dict[str, str] | None) -> str:
+    if not name_map:
+        return native_name
+    return name_map.get(native_name, native_name)
+
+
+def _local_tool_prompts(agent: Any) -> list[tuple[str, str]]:
+    prompt_dirs = subagents.get_paths(agent, "prompts")
+    tool_files = files.get_unique_filenames_in_dirs(
+        prompt_dirs, f"{TOOL_PROMPT_PREFIX}*{TOOL_PROMPT_SUFFIX}"
+    )
+    result: list[tuple[str, str]] = []
+    for tool_file in tool_files:
+        basename = os.path.basename(tool_file)
+        tool_name = _tool_name_from_prompt_basename(basename)
+        if not tool_name:
+            continue
+        try:
+            prompt = agent.read_prompt(basename)
+        except Exception:
+            try:
+                prompt = files.read_file(tool_file)
+            except Exception:
+                prompt = ""
+        result.append((tool_name, prompt))
+    return result
+
+
+def _mcp_tools(agent: Any) -> list[tuple[str, dict[str, Any]]]:
+    try:
+        import helpers.mcp_handler as mcp_helper
+
+        raw_tools = mcp_helper.MCPConfig.get_instance().get_tools()
+    except Exception:
+        return []
+
+    result: list[tuple[str, dict[str, Any]]] = []
+    for entry in raw_tools or []:
+        if not isinstance(entry, dict):
+            continue
+        for tool_name, tool in entry.items():
+            if isinstance(tool, dict):
+                result.append((str(tool_name), tool))
+    return result
+
+
+def _tool_name_from_prompt_basename(basename: str) -> str:
+    if not basename.startswith(TOOL_PROMPT_PREFIX) or not basename.endswith(TOOL_PROMPT_SUFFIX):
+        return ""
+    name = basename[len(TOOL_PROMPT_PREFIX) : -len(TOOL_PROMPT_SUFFIX)]
+    if not name or name in {"tools", "tools_vision"}:
+        return ""
+    return name
+
+
+def _native_tool_name(tool_name: str) -> str:
+    if FUNCTION_NAME_PATTERN.fullmatch(tool_name):
+        return tool_name
+    slug = re.sub(r"[^A-Za-z0-9_-]+", "_", tool_name).strip("_")
+    digest = hashlib.sha1(tool_name.encode("utf-8")).hexdigest()[:8]
+    native = f"{slug[:52]}_{digest}" if slug else f"a0_tool_{digest}"
+    return native[:64]
+
+
+def _description_from_prompt(prompt: str, *, fallback: str) -> str:
+    lines: list[str] = []
+    in_fence = False
+    for raw_line in (prompt or "").splitlines():
+        line = raw_line.strip()
+        if line.startswith("```"):
+            in_fence = not in_fence
+            continue
+        if in_fence or not line:
+            continue
+        if line.startswith("#"):
+            line = line.lstrip("#").strip()
+            if line.lower() == fallback.lower():
+                continue
+        lines.append(line)
+        if sum(len(part) for part in lines) >= MAX_TOOL_DESCRIPTION_CHARS:
+            break
+    description = " ".join(lines).strip() or fallback
+    return _truncate(description)
+
+
+def _schema_from_prompt(prompt: str) -> dict[str, Any]:
+    schema = _schema_from_embedded_json(prompt)
+    if schema:
+        return schema
+    return _schema_from_args_line(prompt)
+
+
+def _schema_from_embedded_json(prompt: str) -> dict[str, Any]:
+    marker = "Input schema for tool_args:"
+    index = (prompt or "").find(marker)
+    if index == -1:
+        return {}
+    tail = prompt[index + len(marker) :].strip()
+    match = re.search(r"\{(?:[^{}]|(?R))*\}", tail, flags=re.DOTALL) if hasattr(re, "VERSION1") else None
+    candidate = match.group(0) if match else _balanced_json_object(tail)
+    if not candidate:
+        return {}
+    try:
+        return _schema_from_any(json.loads(candidate))
+    except Exception:
+        return {}
+
+
+def _schema_from_args_line(prompt: str) -> dict[str, Any]:
+    properties: dict[str, Any] = {}
+    for line in (prompt or "").splitlines():
+        normalized = line.strip()
+        if "args:" not in normalized.lower() and "argument:" not in normalized.lower():
+            continue
+        for name in re.findall(r"`([A-Za-z_][A-Za-z0-9_-]*)`", normalized):
+            properties.setdefault(name, {"type": "string"})
+    if properties:
+        return {
+            "type": "object",
+            "properties": properties,
+            "additionalProperties": True,
+        }
+    return _permissive_schema()
+
+
+def _schema_from_any(schema: Any) -> dict[str, Any]:
+    if isinstance(schema, dict):
+        normalized = dict(schema)
+        normalized.setdefault("type", "object")
+        normalized.setdefault("additionalProperties", True)
+        return normalized
+    return _permissive_schema()
+
+
+def _permissive_schema() -> dict[str, Any]:
+    return {"type": "object", "additionalProperties": True}
+
+
+def _balanced_json_object(text: str) -> str:
+    start = text.find("{")
+    if start == -1:
+        return ""
+    depth = 0
+    in_string = False
+    escape = False
+    for index, char in enumerate(text[start:], start=start):
+        if in_string:
+            if escape:
+                escape = False
+            elif char == "\\":
+                escape = True
+            elif char == '"':
+                in_string = False
+            continue
+        if char == '"':
+            in_string = True
+        elif char == "{":
+            depth += 1
+        elif char == "}":
+            depth -= 1
+            if depth == 0:
+                return text[start : index + 1]
+    return ""
+
+
+def _dedupe_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    seen: set[str] = set()
+    result: list[dict[str, Any]] = []
+    for tool in tools:
+        name = str(tool.get("name") or "")
+        if not name or name in seen:
+            continue
+        seen.add(name)
+        result.append(tool)
+    return result
+
+
+def _truncate(text: str) -> str:
+    if len(text) <= MAX_TOOL_DESCRIPTION_CHARS:
+        return text
+    return text[: MAX_TOOL_DESCRIPTION_CHARS - 3].rstrip() + "..."