add Tool.outputSchema and CallToolResult.structuredContent

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
  "sse-starlette>=1.6.1",
  "pydantic-settings>=2.5.2",
  "uvicorn>=0.23.1; sys_platform != 'emscripten'",
+ "jsonschema==4.23.0",
 ]
 
 [project.optional-dependencies]

src/mcp/client/session.py

@@ -1,8 +1,11 @@
+import logging
+from collections.abc import Awaitable, Callable
 from datetime import timedelta
-from typing import Any, Protocol
+from typing import Any, Protocol, TypeAlias
 
 import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from jsonschema import ValidationError, validate
 from pydantic import AnyUrl, TypeAdapter
 
 import mcp.types as types
@@ -11,6 +14,8 @@
 from mcp.shared.session import BaseSession, ProgressFnT, RequestResponder
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 
+logger = logging.getLogger(__name__)
+
 DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0")
 
 
@@ -44,6 +49,12 @@ async def __call__(
  ) -> None: ...
 
 
+class ToolOutputValidationFnT(Protocol):
+ async def __call__(
+ self, request: types.CallToolRequest, result: types.CallToolResult
+ ) -> bool: ...
+
+
 async def _default_message_handler(
  message: RequestResponder[types.ServerRequest, types.ClientResult]
  | types.ServerNotification
@@ -77,6 +88,25 @@ async def _default_logging_callback(
  pass
 
 
+ToolOutputValidatorProvider: TypeAlias = Callable[
+ ...,
+ Awaitable[ToolOutputValidationFnT],
+]
+
+
+# this bag of spanners is required in order to
+# enable the client session to be parsed to the validator
+async def _python_circularity_hell(arg: Any) -> ToolOutputValidationFnT:
+ # in any sane version of the universe this should never happen
+ # of course in any sane programming language class circularity
+ # dependencies shouldn't be this hard to manage
+ raise RuntimeError(
+ "Help I'm stuck in python circularity hell, please send biscuits"
+ )
+
+
+_default_tool_output_validator: ToolOutputValidatorProvider = _python_circularity_hell
+
 ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(
  types.ClientResult | types.ErrorData
 )
@@ -101,6 +131,7 @@ def __init__(
  logging_callback: LoggingFnT | None = None,
  message_handler: MessageHandlerFnT | None = None,
  client_info: types.Implementation | None = None,
+ tool_output_validator_provider: ToolOutputValidatorProvider | None = None,
  ) -> None:
  super().__init__(
  read_stream,
@@ -114,6 +145,9 @@ def __init__(
  self._list_roots_callback = list_roots_callback or _default_list_roots_callback
  self._logging_callback = logging_callback or _default_logging_callback
  self._message_handler = message_handler or _default_message_handler
+ self._tool_output_validator_provider = (
+ tool_output_validator_provider or _default_tool_output_validator
+ )
 
  async def initialize(self) -> types.InitializeResult:
  sampling = (
@@ -160,6 +194,8 @@ async def initialize(self) -> types.InitializeResult:
  )
  )
 
+ self._tool_output_validator = await self._tool_output_validator_provider(self)
+
  return result
 
  async def send_ping(self) -> types.EmptyResult:
@@ -281,24 +317,33 @@ async def call_tool(
  arguments: dict[str, Any] | None = None,
  read_timeout_seconds: timedelta | None = None,
  progress_callback: ProgressFnT | None = None,
+ validate_result: bool = True,
  ) -> types.CallToolResult:
  """Send a tools/call request with optional progress callback support."""
 
- return await self.send_request(
- types.ClientRequest(
- types.CallToolRequest(
- method="tools/call",
- params=types.CallToolRequestParams(
- name=name,
- arguments=arguments,
- ),
- )
+ request = types.CallToolRequest(
+ method="tools/call",
+ params=types.CallToolRequestParams(
+ name=name,
+ arguments=arguments,
  ),
+ )
+
+ result = await self.send_request(
+ types.ClientRequest(request),
  types.CallToolResult,
  request_read_timeout_seconds=read_timeout_seconds,
  progress_callback=progress_callback,
  )
 
+ if validate_result:
+ valid = await self._tool_output_validator(request, result)
+
+ if not valid:
+ raise RuntimeError("Server responded with invalid result: " f"{result}")
+ # not validating or is valid
+ return result
+
  async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult:
  """Send a prompts/list request."""
  return await self.send_request(
@@ -418,3 +463,75 @@ async def _received_notification(
  await self._logging_callback(params)
  case _:
  pass
+
+
+class NoOpToolOutputValidator(ToolOutputValidationFnT):
+ async def __call__(
+ self, request: types.CallToolRequest, result: types.CallToolResult
+ ) -> bool:
+ return True
+
+
+class SimpleCachingToolOutputValidator(ToolOutputValidationFnT):
+ _schema_cache: dict[str, dict[str, Any] | bool]
+
+ def __init__(self, session: ClientSession):
+ self._session = session
+ self._schema_cache = {}
+ self._refresh_cache = True
+
+ async def __call__(
+ self, request: types.CallToolRequest, result: types.CallToolResult
+ ) -> bool:
+ if result.isError:
+ # allow errors to be propagated
+ return True
+ else:
+ if self._refresh_cache:
+ await self._refresh_schema_cache()
+
+ schema = self._schema_cache.get(request.params.name)
+
+ if schema is None:
+ raise RuntimeError(f"Unknown tool {request.params.name}")
+ elif schema is False:
+ # no schema
+ logging.debug("No schema found checking structuredContent is empty")
+ return result.structuredContent is None
+ else:
+ try:
+ # TODO opportunity to build jsonschema.protocol.Validator
+ # and reuse rather than build every time
+ validate(result.structuredContent, schema)
+ return True
+ except ValidationError as e:
+ logging.exception(e)
+ return False
+
+ async def _refresh_schema_cache(self):
+ cursor = None
+ first = True
+ self._schema_cache = {}
+ while first or cursor is not None:
+ first = False
+ tools_result = await self._session.list_tools(cursor)
+ for tool in tools_result.tools:
+ # store a flag to be able to later distinguish between
+ # no schema for tool and unknown tool which can't be verified
+ schema_or_flag = (
+ False if tool.outputSchema is None else tool.outputSchema
+ )
+ self._schema_cache[tool.name] = schema_or_flag
+ cursor = tools_result.nextCursor
+ continue
+
+ self._refresh_cache = False
+
+
+async def _escape_from_circular_python_hell(
+ session: ClientSession,
+) -> ToolOutputValidationFnT:
+ return SimpleCachingToolOutputValidator(session)
+
+
+_default_tool_output_validator = _escape_from_circular_python_hell

src/mcp/server/fastmcp/server.py

@@ -242,7 +242,7 @@ def run(
  def _setup_handlers(self) -> None:
  """Set up core MCP protocol handlers."""
  self._mcp_server.list_tools()(self.list_tools)
- self._mcp_server.call_tool()(self.call_tool)
+ self._mcp_server.call_tool()(self.call_tool, self._tool_manager.get_schema)
  self._mcp_server.list_resources()(self.list_resources)
  self._mcp_server.read_resource()(self.read_resource)
  self._mcp_server.list_prompts()(self.list_prompts)
@@ -257,6 +257,7 @@ async def list_tools(self) -> list[MCPTool]:
  name=info.name,
  description=info.description,
  inputSchema=info.parameters,
+ outputSchema=info.output,
  annotations=info.annotations,
  )
  for info in tools
@@ -279,7 +280,8 @@ async def call_tool(
  """Call a tool by name with arguments."""
  context = self.get_context()
  result = await self._tool_manager.call_tool(name, arguments, context=context)
- converted_result = _convert_to_content(result)
+ schema = self._tool_manager.get_schema(name)
+ converted_result = _convert_to_content(result, schema)
  return converted_result
 
  async def list_resources(self) -> list[MCPResource]:
@@ -327,6 +329,7 @@ def add_tool(
  name: str | None = None,
  description: str | None = None,
  annotations: ToolAnnotations | None = None,
+ output_schema: dict[str, Any] | None = None,
  ) -> None:
  """Add a tool to the server.
 
@@ -338,9 +341,15 @@ def add_tool(
  name: Optional name for the tool (defaults to function name)
  description: Optional description of what the tool does
  annotations: Optional ToolAnnotations providing additional tool information
+ output_schema: Optional json schema that the tool should output. If
+ not specified the schema will be inferred automatically
  """
  self._tool_manager.add_tool(
- fn, name=name, description=description, annotations=annotations
+ fn,
+ name=name,
+ description=description,
+ annotations=annotations,
+ output_schema=output_schema,
  )
 
  def tool(
@@ -874,25 +883,33 @@ async def get_prompt(
 
 
 def _convert_to_content(
- result: Any,
+ result: Any, schema: dict[str, Any] | None
 ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
- """Convert a result to a sequence of content objects."""
- if result is None:
- return []
-
- if isinstance(result, TextContent | ImageContent | EmbeddedResource):
- return [result]
-
- if isinstance(result, Image):
- return [result.to_image_content()]
-
- if isinstance(result, list | tuple):
- return list(chain.from_iterable(_convert_to_content(item) for item in result)) # type: ignore[reportUnknownVariableType]
+ if schema is None:
+ """Convert a result to a sequence of content objects."""
+ if result is None:
+ return []
+
+ if isinstance(result, TextContent | ImageContent | EmbeddedResource):
+ return [result]
+
+ if isinstance(result, Image):
+ return [result.to_image_content()]
+
+ if isinstance(result, list | tuple):
+ return list(
+ chain.from_iterable(
+ _convert_to_content(item, schema)
+ for item in result # type: ignore[reportUnknownVariableType]
+ )
+ )
 
- if not isinstance(result, str):
- result = pydantic_core.to_json(result, fallback=str, indent=2).decode()
+if not isinstance(result, str):
+result = pydantic_core.to_json(result, fallback=str, indent=2).decode()
 
- return [TextContent(type="text", text=result)]
+ return [TextContent(type="text", text=result)]
+ else:
+ return result
 
 
 class Context(BaseModel, Generic[ServerSessionT, LifespanContextT, RequestT]):

src/mcp/server/fastmcp/tools/base.py

@@ -24,6 +24,10 @@ class Tool(BaseModel):
  name: str = Field(description="Name of the tool")
  description: str = Field(description="Description of what the tool does")
  parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
+ output: dict[str, Any] | None = Field(
+ description="JSON schema for tool output",
+ default=None,
+ )
  fn_metadata: FuncMetadata = Field(
  description="Metadata about the function including a pydantic model for tool"
  " arguments"
@@ -44,6 +48,7 @@ def from_function(
  description: str | None = None,
  context_kwarg: str | None = None,
  annotations: ToolAnnotations | None = None,
+ output_schema: dict[str, Any] | None = None,
  ) -> Tool:
  """Create a Tool from a function."""
  from mcp.server.fastmcp.server import Context
@@ -68,14 +73,17 @@ def from_function(
  func_arg_metadata = func_metadata(
  fn,
  skip_names=[context_kwarg] if context_kwarg is not None else [],
+ output_schema=output_schema,
  )
  parameters = func_arg_metadata.arg_model.model_json_schema()
+ output = func_arg_metadata.output_schema
 
  return cls(
  fn=fn,
  name=func_name,
  description=func_doc,
  parameters=parameters,
+ output=output,
  fn_metadata=func_arg_metadata,
  is_async=is_async,
  context_kwarg=context_kwarg,

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -48,10 +48,15 @@ def add_tool(
  name: str | None = None,
  description: str | None = None,
  annotations: ToolAnnotations | None = None,
+ output_schema: dict[str, Any] | None = None,
  ) -> Tool:
  """Add a tool to the server."""
  tool = Tool.from_function(
- fn, name=name, description=description, annotations=annotations
+ fn,
+ name=name,
+ description=description,
+ annotations=annotations,
+ output_schema=output_schema,
  )
  existing = self._tools.get(tool.name)
  if existing:
@@ -73,3 +78,9 @@ async def call_tool(
  raise ToolError(f"Unknown tool: {name}")
 
  return await tool.run(arguments, context=context)
+
+ def get_schema(self, name: str) -> dict[str, Any] | None:
+ tool = self.get_tool(name)
+ if not tool:
+ raise ToolError(f"Unknown tool: {name}")
+ return tool.output