From 34eb906c2e41b07df077c60fed173615d88a5bd8 Mon Sep 17 00:00:00 2001
From: Ansari Ujale <aujale30@gmail.com>
Date: Fri, 3 Jul 2026 17:47:55 +0530
Subject: [PATCH] fix: extract and map text-embedded tool calls with special
 tokens (#679) - Add a Pydantic efore model validator to the Message class
 schema   to intercept raw server response payloads before object
 instantiation. - Implement robust regular expression string matching to
 isolate embedded   JSON arrays appended with the <|call|> marker token in the
 message text. - Extract tool metadata keys ('task', 'result_name') along with
 target parameters,   reconstruct them into proper Message.ToolCall
 dictionaries, and append   them natively to the 	ool_calls sequence
 array. - Cleanse the user-visible content field by removing backend technical
 signatures   to prevent parameter leakages into the application text UI
 interface. - Implement isolated unit tests validating schema mutations and
 edge-case behaviors.

---
 ollama/_types.py            | 48 +++++++++++++++++++++++++++++++++++++
 tests/test_model_parsing.py | 22 +++++++++++++++++
 2 files changed, 70 insertions(+)
 create mode 100644 tests/test_model_parsing.py

diff --git a/ollama/_types.py b/ollama/_types.py
index 96529d63..34c091a8 100644
--- a/ollama/_types.py
+++ b/ollama/_types.py
@@ -4,6 +4,7 @@
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Mapping, Optional, Sequence, Union
+from pydantic import model_validator 
 
 from pydantic import (
   BaseModel,
@@ -353,6 +354,53 @@ class Function(SubscriptableBaseModel):
   """
   Tools calls to be made by the model.
   """
+  @model_validator(mode='before')
+  @classmethod
+  def handle_embedded_cloud_tool_calls(cls, data: Any) -> Any:
+    """
+    Intercepts raw dictionaries before Pydantic parsing to handle text-embedded tool calls.
+    """
+    import re
+    import json
+
+    if not isinstance(data, dict):
+        return data
+
+    content_str = data.get("content")
+    if content_str and "<|call|>" in content_str:
+        # Match a JSON block {...} right before the special token
+        match = re.search(r'(\{.*?\})<\|call\|>', content_str)
+        
+        if match:
+            try:
+                raw_json_str = match.group(1)
+                parsed_json = json.loads(raw_json_str)
+                
+                # Pull the function name and arguments
+                func_name = parsed_json.get("task") or parsed_json.get("result_name") or "unknown_tool"
+                func_args = parsed_json.get("params", {})
+                
+                # Structure it exactly like the Message.ToolCall format
+                synthesized_tool_call = {
+                    "function": {
+                        "name": func_name,
+                        "arguments": func_args
+                    }
+                }
+                
+                # Ensure the tool_calls list exists and append the new tool
+                if not data.get("tool_calls"):
+                    data["tool_calls"] = []
+                data["tool_calls"].append(synthesized_tool_call)
+                
+                # Update content to hide backend JSON code from the user
+                data["content"] = content_str.replace(match.group(0), "").strip()
+                
+            except json.JSONDecodeError:
+                pass
+
+    return data
+
 
 
 class Tool(SubscriptableBaseModel):
diff --git a/tests/test_model_parsing.py b/tests/test_model_parsing.py
new file mode 100644
index 00000000..fc765817
--- /dev/null
+++ b/tests/test_model_parsing.py
@@ -0,0 +1,22 @@
+import pytest
+from ollama._types import Message
+
+def test_damaged_cloud_output_parsing():
+    """
+    Test that raw text containing text-embedded tool-call structures 
+    is intercepted and parsed correctly into Message objects.
+    """
+
+    simulated_server_payload = {
+        "role": "assistant",
+        "content": '{"params":{"query": "AI research"},"task":"Describe document"}<|call|>commentary'
+    }
+    
+    # This invokes your custom handle_embedded_cloud_tool_calls model validator!
+    message_obj = Message(**simulated_server_payload)
+    assert message_obj.content == "commentary"
+    assert message_obj.tool_calls is not None
+    assert len(message_obj.tool_calls) == 1
+    extracted_tool = message_obj.tool_calls[0]
+    assert extracted_tool.function.name == "Describe document"
+    assert extracted_tool.function.arguments == {"query": "AI research"}