1512edfb85
Four copies of the 40/60 head/tail algorithm with a near-identical notice (terminal_tool_result, mcp_tool_content, code_execution_tool, environments/base_output) collapse into tools/tool_output_truncate.py, so the ratio and the `... [<LABEL> TRUNCATED - N <unit> omitted out of T total] ...` marker are defined once. execute_code keeps byte mode + spill path and only shares the notice/split. Visible change: the terminal notice now uses thousands separators like the other three (`9,000 chars` not `9000 chars`). kanban_specify._truncate: comment claimed escape stripping the body never did; comment now says what the plain clamp is for.
223 lines
11 KiB
Python
223 lines
11 KiB
Python
"""Rendering of MCP tool-result content blocks into model-facing text: size
|
|
capping, _meta filtering, image/audio caching to MEDIA tags, resource links and
|
|
embedded resources."""
|
|
|
|
import base64
|
|
import logging
|
|
import mimetypes
|
|
from typing import Any, Dict, Optional, Tuple
|
|
from tools.ansi_strip import strip_unicode_tags
|
|
from tools.mcp_tool_common import mcp_field
|
|
from tools.mcp_tool_schema import mcp_prefixed_tool_name
|
|
from tools.tool_output_truncate import truncate_head_tail
|
|
|
|
logger = logging.getLogger("tools.mcp_tool")
|
|
|
|
|
|
# Hard ceiling for one MCP text payload (chars), deliberately far ABOVE the budget layer's 50K
|
|
# spillover threshold so ordinary large results reach spillover intact; only floods are lossy.
|
|
# This is the FIRST line of defense against a buggy or malicious MCP server returning multi-megabyte text:
|
|
# without it the full payload is allocated, JSON-encoded and handed downstream before the budget/spillover
|
|
# layer ever sees it (#56059). Distilled from #56060 (Stoltemberg), #56072 (AlexFucuson9) and #56511
|
|
# (Tranquil-Flow), which capped at get_max_bytes() (50K) — correct protection, but at that level it would
|
|
# truncate before spillover could preserve the data. The 40% head / 60% tail split is #56511's shape.
|
|
_MCP_HARD_RESULT_CAP_CHARS = 2_000_000
|
|
# Cap on decoded resource bytes per block (a misbehaving server can't fill the cache disk).
|
|
# Base64 expands ~4/3; oversized payloads are rejected BEFORE decoding (never doubled in memory).
|
|
_MCP_RESOURCE_MAX_BYTES = 50 * 1024 * 1024
|
|
_MCP_RESOURCE_MAX_B64_CHARS = _MCP_RESOURCE_MAX_BYTES * 4 // 3 + 4
|
|
|
|
|
|
def _truncate_mcp_text_result(text: str, max_chars: int = _MCP_HARD_RESULT_CAP_CHARS) -> str:
|
|
"""Bound pathological MCP text before it propagates (#56059)."""
|
|
return truncate_head_tail(text, max_chars, label="MCP RESULT")
|
|
|
|
|
|
def _is_reserved_mcp_meta_key(key: str) -> bool:
|
|
"""True if an MCP ``_meta`` key uses a protocol-reserved prefix: a ``modelcontextprotocol``
|
|
or ``mcp`` label followed by at least one more label. A trailing one
|
|
(``com.example.mcp/...``) is a vendor namespace.
|
|
|
|
Ported from MoonshotAI/kimi-code#2600.
|
|
"""
|
|
slash = key.find("/")
|
|
if slash <= 0:
|
|
return False
|
|
labels = key[:slash].split(".")
|
|
return any(label in ("modelcontextprotocol", "mcp") and i < len(labels) - 1 for i, label in enumerate(labels))
|
|
|
|
|
|
def _strip_reserved_meta_keys(meta) -> Optional[Dict[str, Any]]:
|
|
"""Drop protocol-reserved keys from ``_meta``; None if nothing model-facing remains or the
|
|
input wasn't a mapping."""
|
|
if not isinstance(meta, dict):
|
|
return None
|
|
out = {k: v for k, v in meta.items() if isinstance(k, str) and (not _is_reserved_mcp_meta_key(k))}
|
|
return out or None
|
|
|
|
|
|
def _base_mime(mime_type) -> str:
|
|
"""``type/subtype`` of a MIME string, lower-cased, parameters dropped."""
|
|
return str(mime_type or "").split(";", 1)[0].strip().lower()
|
|
|
|
|
|
def _mcp_image_extension_for_mime_type(mime_type: str) -> str:
|
|
"""File extension for an MCP image MIME type (``.png`` fallback)."""
|
|
normalized = _base_mime(mime_type)
|
|
if normalized in {"image/jpeg", "image/jpg"}:
|
|
return ".jpg"
|
|
return mimetypes.guess_extension(normalized) or ".png"
|
|
|
|
|
|
def _decode_block_b64(data, what: str, label: str, *, cap_what: Optional[str] = None,
|
|
cap_suffix: str = "", decode_fail: str = "") -> Tuple[Optional[bytes], str]:
|
|
"""Base64-decode one block payload: ``(bytes, "")`` or ``(None, inline_marker)``. With
|
|
``cap_what`` the payload is rejected on b64 length BEFORE decoding and on decoded size
|
|
after. Decode failures warn and return ``decode_fail`` ("" = drop the block)."""
|
|
if cap_what and len(data) > _MCP_RESOURCE_MAX_B64_CHARS:
|
|
return None, f"[MCP {cap_what} too large to cache: ~{len(data) * 3 // 4} bytes{cap_suffix}]"
|
|
try:
|
|
raw_bytes = base64.b64decode(data)
|
|
except (TypeError, ValueError) as exc:
|
|
logger.warning("MCP %s decode failed (%s): %s", what, label, exc)
|
|
return None, decode_fail
|
|
if cap_what and len(raw_bytes) > _MCP_RESOURCE_MAX_BYTES:
|
|
return None, f"[MCP {cap_what} too large to cache: {len(raw_bytes)} bytes{cap_suffix}]"
|
|
return raw_bytes, ""
|
|
|
|
|
|
def _write_block_cache(writer: str, what: str, skip_label: str, *args,
|
|
unavailable: str = "", failed: str = "", **kwargs) -> Tuple[Optional[str], str]:
|
|
"""Call ``gateway.platforms.base.<writer>(*args, **kwargs)``: ``(path, "")`` or ``(None,
|
|
marker)``. Fail-open so one bad block never kills the tool result: gateway deps missing
|
|
(cron without gateway) → ``unavailable``; any other cache error → warning + ``failed``."""
|
|
try:
|
|
import gateway.platforms.base as _base
|
|
return getattr(_base, writer)(*args, **kwargs), ""
|
|
except ImportError:
|
|
logger.debug("MCP %s caching skipped — gateway.platforms.base unavailable", skip_label)
|
|
return None, unavailable
|
|
except Exception as exc:
|
|
logger.warning("MCP %s cache failed: %s", what, exc)
|
|
return None, failed
|
|
|
|
|
|
_WAV_MIME_EXT = {"audio/wav": ".wav", "audio/x-wav": ".wav", "audio/wave": ".wav"}
|
|
|
|
|
|
def _cache_mcp_media_block(block, kind: str, writer: str, ext_for, *, cap_what: Optional[str] = None) -> str:
|
|
"""Cache an image/audio block and return a ``MEDIA:<path>`` tag. "" (logging, not raising)
|
|
when the block isn't ``kind`` media, the base64 is malformed, or the cache rejects the
|
|
bytes: the caller falls through to any text blocks."""
|
|
data = getattr(block, "data", None)
|
|
mime = _base_mime(mcp_field(block, "mime_type", "mimeType"))
|
|
if data is None or not mime.startswith(f"{kind}/"):
|
|
return ""
|
|
raw_bytes, err = _decode_block_b64(data, f"{kind} block", mime, cap_what=cap_what)
|
|
if raw_bytes is None:
|
|
return err
|
|
path, err = _write_block_cache(writer, f"{kind} block", kind, raw_bytes, ext=ext_for(mime))
|
|
return err if path is None else f"MEDIA:{path}"
|
|
|
|
|
|
def _cache_mcp_image_block(block) -> str:
|
|
"""Cache an ``ImageContent`` block and return a ``MEDIA:<path>`` tag ("" on any failure)."""
|
|
return _cache_mcp_media_block(block, "image", "cache_image_from_bytes", _mcp_image_extension_for_mime_type)
|
|
|
|
|
|
def _cache_mcp_audio_block(block) -> str:
|
|
"""Cache an ``AudioContent`` block and return a ``MEDIA:<path>`` tag ("" on any failure)."""
|
|
return _cache_mcp_media_block(
|
|
block, "audio", "cache_audio_from_bytes",
|
|
lambda mime: _WAV_MIME_EXT.get(mime) or mimetypes.guess_extension(mime) or ".ogg",
|
|
cap_what="audio resource")
|
|
|
|
|
|
def _mcp_resource_filename(uri: str, mime_type: str) -> str:
|
|
"""Safe display filename from the URI's last path segment, used only as a name hint:
|
|
``cache_document_from_bytes`` re-sanitizes and prefixes it, so remote path components
|
|
can't steer the cache location."""
|
|
import re as _re
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse, unquote
|
|
name = ""
|
|
if uri:
|
|
try:
|
|
name = Path(unquote(urlparse(str(uri)).path or "")).name
|
|
except (ValueError, TypeError):
|
|
pass
|
|
# Strip control chars (hostile URIs could inject newlines/ANSI into the filename and
|
|
# transcript marker) and cap length, preserving the extension.
|
|
name = _re.sub(r"[\x00-\x1f\x7f]", "", name).strip()
|
|
if len(name) > 150:
|
|
stem, dot, ext = name.rpartition(".")
|
|
name = stem[: 150 - len(ext) - 1] + "." + ext if dot and 0 < len(ext) <= 12 else name[:150]
|
|
if not name or name in {".", ".."}:
|
|
ext = mimetypes.guess_extension(_base_mime(mime_type)) or ".bin"
|
|
name = f"resource{ext}"
|
|
return name
|
|
|
|
|
|
def _render_mcp_dropped_block_notice(block, block_type: str) -> str:
|
|
"""Inline notice for an unsupported MCP content block (kimi-code#3227): silently dropping it
|
|
leaves the model unaware content went missing. Carries whatever handles the block exposes —
|
|
mime type, uri, size, name — so the agent can fetch or reason about the missing content."""
|
|
details = [f"type={block_type}"]
|
|
mime = mcp_field(block, "mime_type", "mimeType", None)
|
|
if mime:
|
|
details.append(f"mimeType={mime}")
|
|
uri = getattr(block, "uri", None) or getattr(getattr(block, "resource", None), "uri", None)
|
|
if uri:
|
|
details.append(f"uri={uri}")
|
|
for size_attr in ("size", "sizeInBytes"):
|
|
size = getattr(block, size_attr, None)
|
|
if isinstance(size, int):
|
|
details.append(f"size={size}")
|
|
break
|
|
name = getattr(block, "name", None)
|
|
if name and isinstance(name, str):
|
|
details.append(f"name={name}")
|
|
return f"[MCP content dropped: unsupported block ({', '.join(details)})]"
|
|
|
|
|
|
def _render_mcp_resource_block(block, server_name: str = "") -> str:
|
|
"""Render a ``ResourceLink`` or ``EmbeddedResource`` block as text: embedded text → the
|
|
text; embedded blob → decoded (size-capped) into the document cache with a path marker;
|
|
link → the URI plus a pointer at the server's read_resource tool (no fetch here — links
|
|
are only readable via the originating session). "" for non-resource blocks; failures are
|
|
reported inline rather than silently dropped."""
|
|
block_type = getattr(block, "type", "")
|
|
if block_type == "resource_link" or (hasattr(block, "uri") and not hasattr(block, "resource") and block_type != "text"):
|
|
uri = getattr(block, "uri", None)
|
|
if not uri:
|
|
return ""
|
|
name = getattr(block, "name", "") or ""
|
|
mime = mcp_field(block, "mime_type", "mimeType", "") or ""
|
|
details = f"uri={uri}" + (f", name={name}" if name else "") + (f", mimeType={mime}" if mime else "")
|
|
reader = mcp_prefixed_tool_name(server_name, "read_resource") if server_name else "the MCP server's read_resource tool"
|
|
return f"[MCP resource link: {details} — fetch it with {reader}]"
|
|
resource = getattr(block, "resource", None)
|
|
if resource is None:
|
|
return ""
|
|
text = getattr(resource, "text", None)
|
|
if text is not None:
|
|
return strip_unicode_tags(str(text))
|
|
blob = getattr(resource, "blob", None)
|
|
if blob is None:
|
|
return ""
|
|
uri = str(getattr(resource, "uri", "") or "")
|
|
mime = str(mcp_field(resource, "mime_type", "mimeType", "") or "")
|
|
raw_bytes, err = _decode_block_b64(
|
|
blob, "embedded resource", mime or uri, cap_what="embedded resource", cap_suffix=f", uri={uri}",
|
|
decode_fail=f"[MCP embedded resource could not be decoded: {mime or uri}]")
|
|
if raw_bytes is None:
|
|
return err
|
|
kind = mime or "unknown type"
|
|
path, err = _write_block_cache(
|
|
"cache_document_from_bytes", "embedded resource", "resource", raw_bytes, _mcp_resource_filename(uri, mime),
|
|
unavailable=f"[MCP embedded resource received ({len(raw_bytes)} bytes, {kind}) but document cache unavailable in this process]",
|
|
failed=f"[MCP embedded resource could not be cached: {mime or uri}]")
|
|
if path is None:
|
|
return err
|
|
return f"[MCP resource saved to {path} ({kind}, {len(raw_bytes)} bytes) — read it with read_file or terminal tools]"
|