Module adcp.utils.response_parser

Functions

def parse_json_or_text(data: Any, response_type: type[T]) ‑> ~T

Expand source code

def parse_json_or_text(data: Any, response_type: type[T]) -> T:
    """
    Parse data that might be JSON string, dict, or other format.

    Used by A2A adapter for flexible response parsing.

    Handles protocol-level wrapping where servers return:
    - {"message": "...", "data": {...task_data...}}
    - {"message": "...", ...task_fields...}

    Args:
        data: Response data (string, dict, or other)
        response_type: Expected Pydantic model type

    Returns:
        Parsed and validated response object

    Raises:
        ValueError: If data cannot be parsed into expected type
    """
    # If already a dict, try direct validation
    if isinstance(data, dict):
        # Try direct validation first
        original_error: Exception | None = None
        try:
            return _validate_union_type(data, response_type)
        except (ValidationError, ValueError) as e:
            original_error = e

        # Try extracting task data (separates protocol fields)
        task_data = _extract_task_data(data)
        if task_data is not data:
            try:
                return _validate_union_type(task_data, response_type)
            except (ValidationError, ValueError):
                pass  # Fall through to raise original error

        # Report the original validation error
        type_name = getattr(response_type, "__name__", str(response_type))
        raise ValueError(
            f"Response doesn't match expected schema {type_name}: {original_error}"
        ) from original_error

    # If string, try JSON parsing
    if isinstance(data, str):
        try:
            parsed = json.loads(data)
        except json.JSONDecodeError as e:
            raise ValueError(f"Response is not valid JSON: {e}") from e

        # Recursively handle dict parsing (which includes protocol field extraction)
        if isinstance(parsed, dict):
            return parse_json_or_text(parsed, response_type)

        # Non-dict JSON (shouldn't happen for AdCP responses)
        try:
            return _validate_union_type(parsed, response_type)
        except ValidationError as e:
            type_name = getattr(response_type, "__name__", str(response_type))
            raise ValueError(f"Response doesn't match expected schema {type_name}: {e}") from e

    # Unsupported type
    type_name = getattr(response_type, "__name__", str(response_type))
    raise ValueError(f"Cannot parse response of type {type(data).__name__} into {type_name}")

Parse data that might be JSON string, dict, or other format.

Used by A2A adapter for flexible response parsing.

Handles protocol-level wrapping where servers return: - {"message": "…", "data": {…task_data…}} - {"message": "…", …task_fields…}

Args

data

Response data (string, dict, or other)

response_type

Expected Pydantic model type

Returns

Parsed and validated response object

Raises

ValueError

If data cannot be parsed into expected type

def parse_mcp_content(content: list[dict[str, Any]], response_type: type[T]) ‑> ~T

Expand source code

def parse_mcp_content(content: list[dict[str, Any]], response_type: type[T]) -> T:
    """
    Parse MCP content array into structured response type.

    MCP tools return content as a list of content items:
    [{"type": "text", "text": "..."}, {"type": "resource", ...}]

    The MCP adapter is responsible for serializing MCP SDK Pydantic objects
    to plain dicts before calling this function.

    For AdCP, we expect JSON data in text content items.

    Args:
        content: MCP content array (list of plain dicts)
        response_type: Expected Pydantic model type

    Returns:
        Parsed and validated response object

    Raises:
        ValueError: If content cannot be parsed into expected type
    """
    if not content:
        raise ValueError("Empty MCP content array")

    # Look for text content items that might contain JSON
    for item in content:
        if item.get("type") == "text":
            text = item.get("text", "")
            if not text:
                continue

            try:
                # Try parsing as JSON
                data = json.loads(text)
                # Validate against expected schema (handles Union types)
                return _validate_union_type(data, response_type)
            except json.JSONDecodeError:
                # Not JSON, try next item
                continue
            except ValidationError as e:
                logger.warning(
                    f"MCP content doesn't match expected schema {response_type.__name__}: {e}"
                )
                raise ValueError(f"MCP response doesn't match expected schema: {e}") from e
        elif item.get("type") == "resource":
            # Resource content might have structured data
            try:
                return _validate_union_type(item, response_type)
            except ValidationError:
                # Try next item
                continue

    # If we get here, no content item could be parsed
    # Include content preview for debugging (first 2 items, max 500 chars each)
    content_preview = json.dumps(content[:2], indent=2, default=str)
    if len(content_preview) > 500:
        content_preview = content_preview[:500] + "..."

    raise ValueError(
        f"No valid {response_type.__name__} data found in MCP content. "
        f"Content types: {[item.get('type') for item in content]}. "
        f"Content preview:\n{content_preview}"
    )

Parse MCP content array into structured response type.

MCP tools return content as a list of content items: [{"type": "text", "text": "…"}, {"type": "resource", …}]

The MCP adapter is responsible for serializing MCP SDK Pydantic objects to plain dicts before calling this function.

For AdCP, we expect JSON data in text content items.

Args

content

MCP content array (list of plain dicts)

response_type

Expected Pydantic model type

Returns

Parsed and validated response object

Raises

ValueError

If content cannot be parsed into expected type