81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""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",
|
|
]
|