"""Pure helpers for MCP result envelopes used by the Streamlit console.""" from __future__ import annotations from collections.abc import Mapping from typing import Any def response_payload(value: Any) -> Mapping[str, Any]: """Return the business payload from a direct or nested MCP response.""" if not isinstance(value, Mapping): return {} nested = value.get("response") return nested if isinstance(nested, Mapping) else value def text_result(value: Any) -> str: """Return a textual `result` field without stringifying other structures.""" result = response_payload(value).get("result") return result.strip() if isinstance(result, str) else "" def status_result_summary(value: Any, *, excerpt_chars: int = 900) -> dict[str, Any]: """Build a safe UI summary for status/result-style compatibility tools.""" payload = response_payload(value) summary: dict[str, Any] = {} for key in ("status", "success", "error", "errorCode", "errorMessage"): item = payload.get(key) if item not in (None, "", []): summary[key] = item result = text_result(value) if result: summary["result_chars"] = len(result) summary["result_excerpt"] = result[:excerpt_chars] + ( "..." if len(result) > excerpt_chars else "" ) return summary def status_result_evidence(value: Any, *, max_chars: int = 7000) -> dict[str, Any]: """Preserve bounded textual policy/data evidence for final answer synthesis.""" payload = response_payload(value) evidence: dict[str, Any] = {} for key in ("status", "success", "error", "errorCode", "errorMessage"): item = payload.get(key) if item not in (None, "", []): evidence[key] = item result = text_result(value) if result: evidence["result"] = result[:max_chars] + ( "..." if len(result) > max_chars else "" ) evidence["result_chars"] = len(result) return evidence def has_actionable_text_result(value: Any) -> bool: """Return whether a textual result contains evidence worth stopping on.""" result = text_result(value) if not result: return False normalized = " ".join(result.casefold().split()) return not any( marker in normalized for marker in ("no data found", "no evidence found", "error:") ) __all__ = [ "has_actionable_text_result", "response_payload", "status_result_evidence", "status_result_summary", "text_result", ]