From 50a6c70a2f65a62d45952ba516454639ccce801e Mon Sep 17 00:00:00 2001 From: Aliaksei Venski Date: Fri, 14 Aug 2026 14:48:44 +0200 Subject: [PATCH] fix(mcp): replace Any-typed resource parsing with a validated Pydantic model Greptile flagged Mapping[str, Any] / dict[str, Any] at the parser's external-data boundary. Replace the ad-hoc mapping access with _ResourcePayload, a lenient (extra="allow", all-optional) Pydantic model validated once at the boundary, and _TextContentBlock, a TypedDict for the block this module constructs itself, so neither the parsing nor the construction side needs an explicit Any. The object-typed parameters that must stay maximally generic (arbitrary content blocks that matched no known schema) are already object, not Any, so they are unchanged. --- .../tolerant_result.py | 54 +++++++++++++------ 1 file changed, 38 insertions(+), 16 deletions(-) diff --git a/litellm/experimental_mcp_client/tolerant_result.py b/litellm/experimental_mcp_client/tolerant_result.py index e25612f8a0f..602c317037f 100644 --- a/litellm/experimental_mcp_client/tolerant_result.py +++ b/litellm/experimental_mcp_client/tolerant_result.py @@ -14,12 +14,12 @@ survives. import base64 import json from datetime import timedelta -from typing import Any, Final, Mapping, Sequence +from typing import Any, Final, Literal, Mapping, Sequence, TypedDict from mcp import ClientSession, types from mcp.shared.session import ProgressFnT from mcp.types import CallToolResult as MCPCallToolResult -from pydantic import ValidationError, model_validator +from pydantic import BaseModel, ConfigDict, ValidationError, model_validator def _block_is_valid(block: object) -> bool: @@ -30,27 +30,49 @@ def _block_is_valid(block: object) -> bool: return True -def _resource_text(resource: Mapping[str, Any]) -> str | None: - text: Final = resource.get("text") - if isinstance(text, str): - return text - blob: Final = resource.get("blob") - mime_type: Final = resource.get("mimeType") - if not isinstance(blob, str) or not isinstance(mime_type, str) or not mime_type.startswith("text/"): +class _ResourcePayload(BaseModel): + """Lenient shape of an ``EmbeddedResource.resource``: every field a non-compliant upstream + might have gotten wrong is optional, so this never itself fails to validate.""" + + model_config = ConfigDict(extra="allow") + + text: str | None = None + blob: str | None = None + mimeType: str | None = None # noqa: N815 # mirrors the MCP wire field name + + +class _TextContentBlock(TypedDict): + type: Literal["text"] + text: str + + +def _resource_text(resource: _ResourcePayload) -> str | None: + if resource.text is not None: + return resource.text + if resource.blob is None or resource.mimeType is None or not resource.mimeType.startswith("text/"): return None try: - return base64.b64decode(blob, validate=True).decode("utf-8") + return base64.b64decode(resource.blob, validate=True).decode("utf-8") except (ValueError, UnicodeDecodeError): return None -def _as_text_block(block: object) -> dict[str, Any]: +def _validate_resource(raw_resource: Mapping[str, object]) -> _ResourcePayload | None: + try: + return _ResourcePayload.model_validate(raw_resource) + except ValidationError: + return None + + +def _as_text_block(block: object) -> _TextContentBlock: if isinstance(block, Mapping): - resource: Final = block.get("resource") - if isinstance(resource, Mapping): - text: Final = _resource_text(resource) - if text is not None: - return {"type": "text", "text": text} + raw_resource: Final = block.get("resource") + if isinstance(raw_resource, Mapping): + resource: Final = _validate_resource(raw_resource) + if resource is not None: + text: Final = _resource_text(resource) + if text is not None: + return {"type": "text", "text": text} return {"type": "text", "text": json.dumps(block, default=str)}