ADK ToolCallResult serialization issue. #428
Description
This line explodes when I used it with google's genai db toolbox.
I've captured the ToolCallResult payload.
{'result': CallToolResult(meta=None, content=[TextContent(type='text', text='{"table_name":"","table_schema":""}', annotations=None, meta=None), TextContent(type='text', text='{"table_name":"","table_schema":""}', annotations=None, meta=None), TextContent(type='text', text='{"table_name":"","table_schema":""}', annotations=None, meta=None)], structuredContent=None, isError=False)}
I have a fix locally, but for a PR, I will need to really look through the ag-ui-protocol code itself to fix it properly. No promise, but will be a week quickest if I get around to it.
Steps to reproduce (vaguely, but reliably):
Add a remote tool to the Google Python ADK:
from google.adk.tools import ToolContext, FunctionTool
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
mcp_toolset = McpToolset(
connection_params=StreamableHTTPConnectionParams('
# MCP server endpoint.
url="http://localhost:4500/mcp/"
)
)
Set up the Google Genai Toolbox with local postgres.
sources:
postgres-local:
kind: postgres
host: localhost
port: 5432
database: default
user: root
password: password
tools:
postgres-execute-sql:
kind: postgres-execute-sql
source: postgres-local
description: |-
Use this tool to execute sql statement against a postgresql database
Then tell the agent to use the tool in any way such as "list 3 tables".
My current fix is to monkey patch the json.dump method itself. "Vibe coded" it with AI so excuse the grotesque code.
"""
CallToolResult Serialization Fix
This module provides a targeted fix for the CallToolResult JSON serialization issue
in ag_ui_adk.event_translator at line 380.
"""
import json
import logging
from typing import Any, Dict, List
logger = logging.getLogger("CallToolResultFix")
def serialize_calltooolresult(obj: Any) -> str:
"""
Custom serializer that handles CallToolResult objects and their nested content.
Args:
obj: The object to serialize (typically func_response.response)
Returns:
str: JSON string representation
"""
def convert_to_serializable(item: Any) -> Any:
"""Recursively convert objects to JSON-serializable format."""
# Handle CallToolResult objects
if hasattr(item, '__class__') and item.__class__.__name__ == 'CallToolResult':
result = {
'meta': item.meta,
'isError': item.isError,
'structuredContent': item.structuredContent
}
# Handle content array
if hasattr(item, 'content') and item.content:
content_list = []
for content_item in item.content:
if hasattr(content_item, '__class__') and content_item.__class__.__name__ == 'TextContent':
content_list.append({
'type': content_item.type,
'text': content_item.text,
'annotations': content_item.annotations,
'meta': content_item.meta
})
else:
# Fallback for other content types
content_list.append(str(content_item))
result['content'] = content_list
else:
result['content'] = []
return result
# Handle TextContent objects
elif hasattr(item, '__class__') and item.__class__.__name__ == 'TextContent':
return {
'type': item.type,
'text': item.text,
'annotations': item.annotations,
'meta': item.meta
}
# Handle dictionaries
elif isinstance(item, dict):
return {key: convert_to_serializable(value) for key, value in item.items()}
# Handle lists
elif isinstance(item, list):
return [convert_to_serializable(element) for element in item]
# Handle basic types that are already serializable
elif isinstance(item, (str, int, float, bool, type(None))):
return item
# Fallback: convert to string representation
else:
logger.warning(f"Converting non-serializable object to string: {type(item)}")
return str(item)
try:
serializable_obj = convert_to_serializable(obj)
return json.dumps(serializable_obj)
except Exception as e:
logger.error(f"Failed to serialize object: {e}")
# Ultimate fallback
return json.dumps({"error": f"Serialization failed: {str(e)}", "original_type": str(type(obj))})
def apply_calltooolresult_fix():
"""
Apply the CallToolResult serialization fix by monkey patching json.dumps
to handle CallToolResult objects specifically in the event_translator context.
"""
# Store original json.dumps
original_json_dumps = json.dumps
def patched_json_dumps(obj: Any, *args, **kwargs) -> str:
"""Patched json.dumps that handles CallToolResult serialization."""
try:
return original_json_dumps(obj, *args, **kwargs)
except TypeError as e:
if "CallToolResult" in str(e):
logger.info("🔧 Applying CallToolResult serialization fix")
return serialize_calltooolresult(obj)
else:
# Re-raise for other TypeError cases
raise
# Apply the patch
json.dumps = patched_json_dumps
logger.info("✅ CallToolResult serialization fix applied")
return True
if __name__ == "__main__":
apply_calltooolresult_fix()
Then just call apply_calltooolresult_fix() somewhere to apply it.
@contextablemark FYI.