holetron-lab/fleet-memory
1
0
Fork 0
Code Releases Packages Activity Actions

fix(mcp): auto-coerce string-encoded JSON in tool arguments (#849) (#899)

Browse source 
LLM agents frequently serialize list/dict tool arguments as JSON strings
instead of native types (e.g., tags='["a","b"]' instead of tags=["a","b"]),
causing Pydantic validation failures. This extends _make_tools_tolerant to
detect array/object parameters from the JSON Schema and auto-coerce string
values via json.loads before validation.

Also fixes _make_tools_tolerant compatibility with FastMCP 3.x by adding
a _get_mcp_tools helper that supports both 2.x and 3.x internal APIs.
This commit is contained in:
Nicolò Boschi 2026-04-07 09:33:12 +02:00 committed by GitHub
parent 26794aab09
commit 443c94c827
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 429 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
hindsight-api-slim/hindsight_api/api/mcp.py
View file

@ -156,24 +156,65 @@ def create_mcp_server(memory: MemoryEngine, multi_bank: bool = True) -> FastMCP:
return mcp
def _get_mcp_tools(mcp: FastMCP) -> dict:
"""Get tool name→object mapping, compatible with FastMCP 2.x and 3.x."""
# FastMCP 2.x: _tool_manager._tools
if hasattr(mcp, "_tool_manager"):
return mcp._tool_manager._tools # type: ignore[union-attr]
# FastMCP 3.x: _local_provider._components with "tool:" prefix
if hasattr(mcp, "_local_provider"):
return {
k.split(":")[1].split("@")[0]: v
for k, v in mcp._local_provider._components.items() # type: ignore[union-attr]
if k.startswith("tool:")
}
msg = "Cannot locate tools on FastMCP instance"
raise AttributeError(msg)
def _make_tools_tolerant(mcp: FastMCP) -> None:
"""Wrap all tool run methods to strip unknown arguments before validation.
"""Wrap all tool run methods to strip unknown arguments and coerce string-encoded JSON.
LLMs frequently add extra fields like "explanation" or "reasoning" to tool calls.
FastMCP's Pydantic TypeAdapter rejects these with "Unexpected keyword argument".
This wraps each tool's run() to filter arguments to only known parameters.
LLMs also frequently serialize list/dict arguments as JSON strings instead of native
types (e.g., tags='["a","b"]' instead of tags=["a","b"]). This auto-coerces them.
This wraps each tool's run() to apply both fixes before validation.
"""
try:
for name, tool in mcp._tool_manager._tools.items(): # type: ignore[unresolved-attribute] # FastMCP 2.x internal; guarded by try/except
tools = _get_mcp_tools(mcp)
for name, tool in tools.items():
if hasattr(tool, "parameters") and tool.parameters:
allowed = set(tool.parameters.get("properties", {}).keys())
properties = tool.parameters.get("properties", {})
allowed = set(properties.keys())
# Build sets of parameter names that expect array or object types.
# Handles both direct types {"type": "array"} and anyOf/oneOf unions
# like {"anyOf": [{"type": "array", ...}, {"type": "null"}]}.
array_params: set[str] = set()
object_params: set[str] = set()
for param_name, param_schema in properties.items():
_collect_coercible_types(param_schema, param_name, array_params, object_params)
original_run = tool.run
async def _tolerant_run(arguments, _allowed=allowed, _orig=original_run):
async def _tolerant_run(
arguments,
_allowed=allowed,
_orig=original_run,
_array_params=array_params,
_object_params=object_params,
):
extra_keys = set(arguments.keys()) - _allowed
if extra_keys:
logger.debug(f"Stripping unknown arguments from tool call: {extra_keys}")
arguments = {k: v for k, v in arguments.items() if k in _allowed}
# Coerce string-encoded JSON for list/dict parameters
arguments = _coerce_string_json(arguments, _array_params, _object_params)
return await _orig(arguments)
# FunctionTool is a Pydantic model with extra='forbid', so use
@ -183,6 +224,59 @@ def _make_tools_tolerant(mcp: FastMCP) -> None:
logger.warning(f"Could not make tools tolerant of extra arguments: {e}")
def _collect_coercible_types(schema: dict, param_name: str, array_params: set[str], object_params: set[str]) -> None:
"""Check a JSON Schema property and add param_name to array_params/object_params if applicable."""
# Direct type
schema_type = schema.get("type")
if schema_type == "array":
array_params.add(param_name)
return
if schema_type == "object":
object_params.add(param_name)
return
# anyOf / oneOf unions (e.g., list[str] | None → {"anyOf": [{"type": "array"}, {"type": "null"}]})
for variant in schema.get("anyOf", []) + schema.get("oneOf", []):
variant_type = variant.get("type")
if variant_type == "array":
array_params.add(param_name)
return
if variant_type == "object":
object_params.add(param_name)
return
def _coerce_string_json(arguments: dict, array_params: set[str], object_params: set[str]) -> dict:
"""Auto-coerce string-encoded JSON arrays/objects to native types.
LLM agents frequently serialize list and dict tool arguments as JSON strings.
This is backward-compatible: native arrays/objects pass through unchanged.
"""
for param_name in array_params:
val = arguments.get(param_name)
if isinstance(val, str):
try:
parsed = json.loads(val)
if isinstance(parsed, list):
arguments = {**arguments, param_name: parsed}
logger.debug(f"Coerced string to list for parameter '{param_name}'")
except (json.JSONDecodeError, TypeError):
pass
for param_name in object_params:
val = arguments.get(param_name)
if isinstance(val, str):
try:
parsed = json.loads(val)
if isinstance(parsed, dict):
arguments = {**arguments, param_name: parsed}
logger.debug(f"Coerced string to dict for parameter '{param_name}'")
except (json.JSONDecodeError, TypeError):
pass
return arguments
class MCPMiddleware:
"""ASGI middleware that intercepts MCP requests and routes to appropriate MCP server.

330
hindsight-api-slim/tests/test_mcp_string_coercion.py Normal file
View file

@ -0,0 +1,330 @@
"""Tests for MCP tool argument string-to-JSON coercion (issue #849)."""
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from hindsight_api.api.mcp import (
_coerce_string_json,
_collect_coercible_types,
_get_mcp_tools,
_make_tools_tolerant,
)
# ---------------------------------------------------------------------------
# _collect_coercible_types — schema type detection
# ---------------------------------------------------------------------------
class TestCollectCoercibleTypes:
"""Tests for _collect_coercible_types schema detection."""
def _run(self, schema: dict, param_name: str = "p") -> tuple[set[str], set[str]]:
array_params: set[str] = set()
object_params: set[str] = set()
_collect_coercible_types(schema, param_name, array_params, object_params)
return array_params, object_params
# --- array types ---
def test_direct_array_type(self):
arrays, objects = self._run({"type": "array", "items": {"type": "string"}})
assert "p" in arrays and not objects
def test_anyof_nullable_array(self):
"""list[str] | None → anyOf with array and null."""
arrays, objects = self._run(
{"anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}]}
)
assert "p" in arrays
def test_oneof_nullable_array(self):
"""oneOf variant."""
arrays, objects = self._run(
{"oneOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}]}
)
assert "p" in arrays
# --- object types ---
def test_direct_object_type(self):
arrays, objects = self._run({"type": "object"})
assert "p" in objects and not arrays
def test_anyof_nullable_object(self):
"""dict[str, str] | None → anyOf with object and null."""
arrays, objects = self._run({"anyOf": [{"type": "object"}, {"type": "null"}]})
assert "p" in objects
def test_oneof_nullable_object(self):
arrays, objects = self._run({"oneOf": [{"type": "object"}, {"type": "null"}]})
assert "p" in objects
# --- non-coercible types (should be ignored) ---
def test_string_type_ignored(self):
arrays, objects = self._run({"type": "string"})
assert not arrays and not objects
def test_integer_type_ignored(self):
arrays, objects = self._run({"type": "integer"})
assert not arrays and not objects
def test_number_type_ignored(self):
arrays, objects = self._run({"type": "number"})
assert not arrays and not objects
def test_boolean_type_ignored(self):
arrays, objects = self._run({"type": "boolean"})
assert not arrays and not objects
def test_null_type_ignored(self):
arrays, objects = self._run({"type": "null"})
assert not arrays and not objects
def test_anyof_string_or_null_ignored(self):
"""str | None should not be collected."""
arrays, objects = self._run({"anyOf": [{"type": "string"}, {"type": "null"}]})
assert not arrays and not objects
def test_anyof_integer_or_null_ignored(self):
arrays, objects = self._run({"anyOf": [{"type": "integer"}, {"type": "null"}]})
assert not arrays and not objects
# ---------------------------------------------------------------------------
# _coerce_string_json — value coercion
# ---------------------------------------------------------------------------
class TestCoerceStringJson:
"""Tests for _coerce_string_json argument coercion."""
# --- list coercion ---
def test_coerce_string_to_list(self):
result = _coerce_string_json(
{"tags": '["tag1", "tag2"]', "query": "hello"},
array_params={"tags"},
object_params=set(),
)
assert result["tags"] == ["tag1", "tag2"]
assert result["query"] == "hello"
def test_coerce_empty_list_string(self):
result = _coerce_string_json({"tags": "[]"}, array_params={"tags"}, object_params=set())
assert result["tags"] == []
def test_native_list_passthrough(self):
result = _coerce_string_json({"tags": ["a", "b"]}, array_params={"tags"}, object_params=set())
assert result["tags"] == ["a", "b"]
# --- dict coercion ---
def test_coerce_string_to_dict(self):
result = _coerce_string_json(
{"metadata": '{"key": "value"}'},
array_params=set(),
object_params={"metadata"},
)
assert result["metadata"] == {"key": "value"}
def test_coerce_empty_dict_string(self):
result = _coerce_string_json({"metadata": "{}"}, array_params=set(), object_params={"metadata"})
assert result["metadata"] == {}
def test_native_dict_passthrough(self):
result = _coerce_string_json(
{"metadata": {"key": "value"}}, array_params=set(), object_params={"metadata"}
)
assert result["metadata"] == {"key": "value"}
# --- non-coercible values left untouched ---
def test_none_passthrough(self):
result = _coerce_string_json({"tags": None}, array_params={"tags"}, object_params=set())
assert result["tags"] is None
def test_invalid_json_string_passthrough(self):
result = _coerce_string_json({"tags": "not-json"}, array_params={"tags"}, object_params=set())
assert result["tags"] == "not-json"
def test_wrong_json_type_not_coerced_list(self):
"""String that parses to a dict should NOT be coerced for an array param."""
result = _coerce_string_json(
{"tags": '{"key": "value"}'}, array_params={"tags"}, object_params=set()
)
assert result["tags"] == '{"key": "value"}'
def test_wrong_json_type_not_coerced_dict(self):
"""String that parses to a list should NOT be coerced for an object param."""
result = _coerce_string_json(
{"metadata": '["a", "b"]'}, array_params=set(), object_params={"metadata"}
)
assert result["metadata"] == '["a", "b"]'
def test_string_param_not_touched(self):
"""Strings not in array_params/object_params are never modified."""
result = _coerce_string_json(
{"query": '["looks", "like", "json"]'},
array_params=set(),
object_params=set(),
)
assert result["query"] == '["looks", "like", "json"]'
def test_integer_param_not_touched(self):
result = _coerce_string_json(
{"max_tokens": 4096}, array_params=set(), object_params=set()
)
assert result["max_tokens"] == 4096
def test_boolean_param_not_touched(self):
result = _coerce_string_json(
{"verbose": True}, array_params=set(), object_params=set()
)
assert result["verbose"] is True
def test_missing_param_no_error(self):
result = _coerce_string_json(
{"query": "hello"},
array_params={"tags"},
object_params={"metadata"},
)
assert result == {"query": "hello"}
# --- multiple params coerced at once ---
def test_multiple_params_coerced(self):
result = _coerce_string_json(
{
"tags": '["a", "b"]',
"types": '["world"]',
"metadata": '{"source": "test"}',
"query": "hello",
"max_tokens": 4096,
},
array_params={"tags", "types"},
object_params={"metadata"},
)
assert result["tags"] == ["a", "b"]
assert result["types"] == ["world"]
assert result["metadata"] == {"source": "test"}
assert result["query"] == "hello"
assert result["max_tokens"] == 4096
# ---------------------------------------------------------------------------
# _make_tools_tolerant — integration test with a real FastMCP tool
# ---------------------------------------------------------------------------
class TestMakeToolsTolerantIntegration:
"""Test that _make_tools_tolerant correctly wraps real FastMCP tool functions."""
def _create_mcp_with_tool(self):
"""Create a FastMCP instance with a tool that uses various parameter types."""
from fastmcp import FastMCP
mcp = FastMCP("test")
captured = {}
@mcp.tool(description="test tool with diverse param types")
async def test_tool(
query: str,
max_tokens: int = 100,
verbose: bool = False,
tags: list[str] | None = None,
metadata: dict[str, str] | None = None,
) -> dict:
"""Test tool.
Args:
query: a string param
max_tokens: an integer param
verbose: a boolean param
tags: an array param
metadata: an object param
"""
captured["query"] = query
captured["max_tokens"] = max_tokens
captured["verbose"] = verbose
captured["tags"] = tags
captured["metadata"] = metadata
return {"ok": True}
return mcp, captured
@pytest.mark.asyncio
async def test_coerces_string_encoded_list(self):
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({"query": "hi", "tags": '["a", "b"]'})
assert captured["tags"] == ["a", "b"]
@pytest.mark.asyncio
async def test_coerces_string_encoded_dict(self):
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({"query": "hi", "metadata": '{"k": "v"}'})
assert captured["metadata"] == {"k": "v"}
@pytest.mark.asyncio
async def test_native_types_pass_through(self):
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({
"query": "hi",
"max_tokens": 200,
"verbose": True,
"tags": ["x"],
"metadata": {"a": "b"},
})
assert captured["query"] == "hi"
assert captured["max_tokens"] == 200
assert captured["verbose"] is True
assert captured["tags"] == ["x"]
assert captured["metadata"] == {"a": "b"}
@pytest.mark.asyncio
async def test_strips_extra_args_and_coerces(self):
"""Both extra-arg stripping and coercion work together."""
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({
"query": "hi",
"tags": '["x"]',
"explanation": "LLM added this",
})
assert captured["tags"] == ["x"]
assert "explanation" not in captured
@pytest.mark.asyncio
async def test_string_param_not_coerced(self):
"""A string param whose value happens to look like JSON is NOT coerced."""
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({"query": '["this", "is", "a", "string"]'})
assert captured["query"] == '["this", "is", "a", "string"]'
@pytest.mark.asyncio
async def test_integer_param_not_coerced(self):
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({"query": "hi", "max_tokens": 50})
assert captured["max_tokens"] == 50
@pytest.mark.asyncio
async def test_boolean_param_not_coerced(self):
mcp, captured = self._create_mcp_with_tool()
_make_tools_tolerant(mcp)
tool = _get_mcp_tools(mcp)["test_tool"]
await tool.run({"query": "hi", "verbose": True})
assert captured["verbose"] is True