Implement client-side argument serialization with focused tests

- Add pydantic_core.to_json() serialization for non-string prompt arguments
- Update type annotations to accept dict[str, Any] instead of dict[str, str]
- Add focused tests covering specific scenarios:
  * Client always serializes non-string args regardless of server types
  * Integration with server-side type conversion
  * Client serialization error with specific PydanticSerializationError
  * Server deserialization error with specific McpError match

This ensures MCP protocol compliance while maintaining developer experience
with typed arguments.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin 2025-06-22 10:18:48 -04:00
commit a7f14d90a4
2 changed files with 112 additions and 5 deletions

View file

@ -7,6 +7,7 @@ from typing import Any, Generic, Literal, cast, overload
import anyio
import httpx
import mcp.types
import pydantic_core
from exceptiongroup import catch
from mcp import ClientSession
from pydantic import AnyUrl
@ -508,13 +509,13 @@ class Client(Generic[ClientTransportT]):
# --- Prompt ---
async def get_prompt_mcp(
self, name: str, arguments: dict[str, str] | None = None
self, name: str, arguments: dict[str, Any] | None = None
) -> mcp.types.GetPromptResult:
"""Send a prompts/get request and return the complete MCP protocol result.
Args:
name (str): The name of the prompt to retrieve.
arguments (dict[str, str] | None, optional): Arguments to pass to the prompt. Defaults to None.
arguments (dict[str, Any] | None, optional): Arguments to pass to the prompt. Defaults to None.
Returns:
mcp.types.GetPromptResult: The complete response object from the protocol,
@ -523,17 +524,30 @@ class Client(Generic[ClientTransportT]):
Raises:
RuntimeError: If called while the client is not connected.
"""
result = await self.session.get_prompt(name=name, arguments=arguments)
# Serialize arguments for MCP protocol - convert non-string values to JSON
serialized_arguments: dict[str, str] | None = None
if arguments:
serialized_arguments = {}
for key, value in arguments.items():
if isinstance(value, str):
serialized_arguments[key] = value
else:
# Use pydantic_core.to_json for consistent serialization
serialized_arguments[key] = pydantic_core.to_json(value).decode()
result = await self.session.get_prompt(
name=name, arguments=serialized_arguments
)
return result
async def get_prompt(
self, name: str, arguments: dict[str, str] | None = None
self, name: str, arguments: dict[str, Any] | None = None
) -> mcp.types.GetPromptResult:
"""Retrieve a rendered prompt message list from the server.
Args:
name (str): The name of the prompt to retrieve.
arguments (dict[str, str] | None, optional): Arguments to pass to the prompt. Defaults to None.
arguments (dict[str, Any] | None, optional): Arguments to pass to the prompt. Defaults to None.
Returns:
mcp.types.GetPromptResult: The complete response object from the protocol,

View file

@ -220,6 +220,99 @@ async def test_get_prompt_mcp(fastmcp_server):
assert result.description == "Example greeting prompt."
async def test_client_serializes_all_non_string_arguments():
"""Test that client always serializes non-string arguments to JSON, regardless of server types."""
server = FastMCP("TestServer")
@server.prompt
def echo_args(arg1: str, arg2: str, arg3: str) -> str:
"""Server accepts all string args but client sends mixed types."""
return f"arg1: {arg1}, arg2: {arg2}, arg3: {arg3}"
client = Client(transport=FastMCPTransport(server))
async with client:
result = await client.get_prompt(
"echo_args",
{
"arg1": "hello", # string - should pass through
"arg2": [1, 2, 3], # list - should be JSON serialized
"arg3": {"key": "value"}, # dict - should be JSON serialized
},
)
content = result.messages[0].content.text # type: ignore[attr-defined]
assert "arg1: hello" in content
assert "arg2: [1,2,3]" in content # JSON serialized list
assert 'arg3: {"key":"value"}' in content # JSON serialized dict
async def test_client_server_type_conversion_integration():
"""Test that client serialization works with server-side type conversion."""
server = FastMCP("TestServer")
@server.prompt
def typed_prompt(numbers: list[int], config: dict[str, str]) -> str:
"""Server expects typed args - will convert from JSON strings."""
return f"Got {len(numbers)} numbers and {len(config)} config items"
client = Client(transport=FastMCPTransport(server))
async with client:
result = await client.get_prompt(
"typed_prompt",
{"numbers": [1, 2, 3, 4], "config": {"theme": "dark", "lang": "en"}},
)
content = result.messages[0].content.text # type: ignore[attr-defined]
assert "Got 4 numbers and 2 config items" in content
async def test_client_serialization_error():
"""Test client error when object cannot be serialized."""
import pydantic_core
server = FastMCP("TestServer")
@server.prompt
def any_prompt(data: str) -> str:
return f"Got: {data}"
# Create an unserializable object
class UnserializableClass:
def __init__(self):
self.func = lambda x: x # functions can't be JSON serialized
client = Client(transport=FastMCPTransport(server))
async with client:
with pytest.raises(pydantic_core.PydanticSerializationError, match="Unable to serialize"):
await client.get_prompt("any_prompt", {"data": UnserializableClass()})
async def test_server_deserialization_error():
"""Test server error when JSON string cannot be converted to expected type."""
from mcp import McpError
server = FastMCP("TestServer")
@server.prompt
def strict_typed_prompt(numbers: list[int]) -> str:
"""Expects list of integers but will receive invalid JSON."""
return f"Got {len(numbers)} numbers"
client = Client(transport=FastMCPTransport(server))
async with client:
with pytest.raises(McpError, match="Error rendering prompt"):
await client.get_prompt(
"strict_typed_prompt",
{
"numbers": "not valid json" # This will fail server-side conversion
},
)
async def test_read_resource_invalid_uri(fastmcp_server):
"""Test reading a resource with an invalid URI."""
client = Client(transport=FastMCPTransport(fastmcp_server))