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:
                type_name = getattr(response_type, "__name__", str(response_type))
                logger.warning(
                    f"MCP content doesn't match expected schema {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] + "..."

    type_name = getattr(response_type, "__name__", str(response_type))
    raise ValueError(
        f"No valid {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