diff --git a/nanobot/providers/openai_compat_provider.py b/nanobot/providers/openai_compat_provider.py
index 57eccdb6..824e7748 100644
--- a/nanobot/providers/openai_compat_provider.py
+++ b/nanobot/providers/openai_compat_provider.py
@@ -7,6 +7,7 @@ import hashlib
import importlib.util
import json
import os
+import re
import secrets
import string
import time
@@ -70,6 +71,7 @@ _KIMI_ALWAYS_THINKING_MODELS: frozenset[str] = frozenset({
"kimi-k2.7-code",
"kimi-k2.7-code-highspeed",
})
+_TEXT_TOOL_CALL_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL)
# Thinking-capable MiMo models per Xiaomi docs (see
# tests/providers/test_xiaomi_mimo_thinking.py). mimo-v2-flash is omitted
# because it does not support thinking.
@@ -165,6 +167,62 @@ def _short_tool_id() -> str:
return "".join(secrets.choice(_ALNUM) for _ in range(9))
+def _strip_json_fence(text: str) -> str:
+ stripped = text.strip()
+ if not stripped.startswith("```") or not stripped.endswith("```"):
+ return stripped
+ lines = stripped.splitlines()
+ if len(lines) < 2:
+ return stripped
+ return "\n".join(lines[1:-1]).strip()
+
+
+def _extract_text_tool_calls(content: str | None) -> tuple[str | None, list[ToolCallRequest]]:
+ """Normalize common text-format tool call blocks into structured calls."""
+ if not content or "" not in content:
+ return content, []
+
+ tool_calls: list[ToolCallRequest] = []
+ spans: list[tuple[int, int]] = []
+ for match in _TEXT_TOOL_CALL_RE.finditer(content):
+ try:
+ payload = json.loads(_strip_json_fence(match.group(1)))
+ except Exception:
+ continue
+ if not isinstance(payload, dict):
+ continue
+
+ nested = payload.get("tool_call")
+ if isinstance(nested, dict):
+ payload = nested
+ function = payload.get("function")
+ if not isinstance(function, dict):
+ function = payload
+ name = function.get("name")
+ if not isinstance(name, str) or not name:
+ continue
+
+ arguments = function.get("arguments", payload.get("arguments", {}))
+ tool_calls.append(ToolCallRequest(
+ id=str(payload.get("id") or _short_tool_id()),
+ name=name,
+ arguments=parse_tool_arguments(arguments),
+ ))
+ spans.append(match.span())
+
+ if not tool_calls:
+ return content, []
+
+ visible_parts: list[str] = []
+ last = 0
+ for start, end in spans:
+ visible_parts.append(content[last:start])
+ last = end
+ visible_parts.append(content[last:])
+ visible_content = "".join(visible_parts).strip() or None
+ return visible_content, tool_calls
+
+
def _get(obj: Any, key: str) -> Any:
"""Get a value from dict or object attribute, returning None if absent."""
if isinstance(obj, dict):
@@ -1161,6 +1219,8 @@ class OpenAICompatProvider(LLMProvider):
provider_specific_fields=prov,
function_provider_specific_fields=fn_prov,
))
+ if not parsed_tool_calls:
+ content, parsed_tool_calls = _extract_text_tool_calls(content)
return LLMResponse(
content=content,
@@ -1206,6 +1266,8 @@ class OpenAICompatProvider(LLMProvider):
provider_specific_fields=prov,
function_provider_specific_fields=fn_prov,
))
+ if not tool_calls:
+ content, tool_calls = _extract_text_tool_calls(content)
reasoning_content = getattr(msg, "reasoning_content", None)
if reasoning_content is None and getattr(msg, "reasoning", None):
@@ -1343,19 +1405,24 @@ class OpenAICompatProvider(LLMProvider):
b["id"] = _short_tool_id()
_seen_tc_ids.add(b["id"])
+ content = "".join(content_parts) or None
+ tool_calls = [
+ ToolCallRequest(
+ id=b["id"] or _short_tool_id(),
+ name=b["name"],
+ arguments=parse_tool_arguments(b["arguments"]),
+ extra_content=b.get("extra_content"),
+ provider_specific_fields=b.get("prov"),
+ function_provider_specific_fields=b.get("fn_prov"),
+ )
+ for b in tc_bufs.values()
+ ]
+ if not tool_calls:
+ content, tool_calls = _extract_text_tool_calls(content)
+
return LLMResponse(
- content="".join(content_parts) or None,
- tool_calls=[
- ToolCallRequest(
- id=b["id"] or _short_tool_id(),
- name=b["name"],
- arguments=parse_tool_arguments(b["arguments"]),
- extra_content=b.get("extra_content"),
- provider_specific_fields=b.get("prov"),
- function_provider_specific_fields=b.get("fn_prov"),
- )
- for b in tc_bufs.values()
- ],
+ content=content,
+ tool_calls=tool_calls,
finish_reason=finish_reason,
usage=usage,
reasoning_content="".join(reasoning_parts) or None,
diff --git a/tests/providers/test_custom_provider.py b/tests/providers/test_custom_provider.py
index ee1f9a09..02388897 100644
--- a/tests/providers/test_custom_provider.py
+++ b/tests/providers/test_custom_provider.py
@@ -49,6 +49,51 @@ def test_custom_provider_parse_accepts_dict_response() -> None:
assert result.usage["total_tokens"] == 3
+def test_custom_provider_parse_normalizes_text_tool_call() -> None:
+ with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
+ provider = OpenAICompatProvider()
+
+ result = provider._parse({
+ "choices": [{
+ "message": {
+ "content": (
+ "I'll inspect it.\n"
+ '{"name":"read_file","arguments":{"path":"README.md"}}'
+ ""
+ ),
+ },
+ "finish_reason": "stop",
+ }],
+ })
+
+ assert result.content == "I'll inspect it."
+ assert len(result.tool_calls) == 1
+ assert result.tool_calls[0].name == "read_file"
+ assert result.tool_calls[0].arguments == {"path": "README.md"}
+
+
+def test_custom_provider_parse_keeps_structured_tool_call_over_text_markup() -> None:
+ with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
+ provider = OpenAICompatProvider()
+
+ result = provider._parse({
+ "choices": [{
+ "message": {
+ "content": '{"name":"ignored","arguments":{}}',
+ "tool_calls": [{
+ "id": "call_structured",
+ "function": {"name": "list_dir", "arguments": '{"path":"."}'},
+ }],
+ },
+ "finish_reason": "tool_calls",
+ }],
+ })
+
+ assert len(result.tool_calls) == 1
+ assert result.tool_calls[0].id == "call_structured"
+ assert result.tool_calls[0].name == "list_dir"
+
+
def test_custom_provider_parse_chunks_accepts_plain_text_chunks() -> None:
result = OpenAICompatProvider._parse_chunks(["hello ", "world"])
@@ -56,6 +101,22 @@ def test_custom_provider_parse_chunks_accepts_plain_text_chunks() -> None:
assert result.content == "hello world"
+def test_custom_provider_parse_chunks_normalizes_split_text_tool_call() -> None:
+ chunks = [
+ {"choices": [{"delta": {"content": ""}}]},
+ {"choices": [{"delta": {"content": '{"name":"list_dir",'}}]},
+ {"choices": [{"delta": {"content": '"arguments":{"path":"."}}'}}]},
+ {"choices": [{"finish_reason": "stop", "delta": {}}]},
+ ]
+
+ result = OpenAICompatProvider._parse_chunks(chunks)
+
+ assert result.content is None
+ assert len(result.tool_calls) == 1
+ assert result.tool_calls[0].name == "list_dir"
+ assert result.tool_calls[0].arguments == {"path": "."}
+
+
def test_custom_provider_parse_chunks_deduplicates_parallel_tool_call_ids() -> None:
chunks = [{
"choices": [{