"""
Transformation for Bedrock AgentCore

https://docs.aws.amazon.com/bedrock/latest/APIReference/API_agentcore_InvokeAgentRuntime.html
"""

import json
from collections.abc import AsyncGenerator
from typing import TYPE_CHECKING, Any, Final, Optional, Union, cast
from urllib.parse import quote

import httpx

from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.prompt_templates.common_utils import (
    convert_content_list_to_str,
)
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.llms.a2a.common_utils import extract_text_from_a2a_response
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.types.llms.bedrock_agentcore import (
    AgentCoreMessage,
    AgentCoreParsedResponse,
    AgentCoreUsage,
)
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import (
    Choices,
    Delta,
    Message,
    ModelResponse,
    ModelResponseStream,
    StreamingChoices,
    Usage,
)

if TYPE_CHECKING:
    import tiktoken

    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
    from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler

    LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
    LiteLLMLoggingObj = Any
    HTTPHandler = Any
    AsyncHTTPHandler = Any


class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM):
    def __init__(self, **kwargs):
        BaseConfig.__init__(self, **kwargs)
        BaseAWSLLM.__init__(self, **kwargs)

    def get_supported_openai_params(self, model: str) -> list[str]:
        """
        Bedrock AgentCore has 0 OpenAI compatible params
        """
        return []

    def map_openai_params(
        self,
        non_default_params: dict,
        optional_params: dict,
        model: str,
        drop_params: bool,
    ) -> dict:
        """
        Map OpenAI params to AgentCore params
        """
        return optional_params

    def get_complete_url(
        self,
        api_base: str | None,
        api_key: str | None,
        model: str,
        optional_params: dict,
        litellm_params: dict,
        stream: bool | None = None,
    ) -> str:
        """
        Get the complete url for the request
        """
        ### SET RUNTIME ENDPOINT ###
        aws_bedrock_runtime_endpoint: Final = optional_params.get("aws_bedrock_runtime_endpoint", None)

        # Extract ARN from model string
        agent_runtime_arn: Final = self._get_agent_runtime_arn(model)

        # Parse ARN to get region
        region: Final = self._extract_region_from_arn(agent_runtime_arn)

        # Build the base endpoint URL for AgentCore
        # Note: We don't use get_runtime_endpoint as AgentCore has its own endpoint structure
        if aws_bedrock_runtime_endpoint:
            base_url = aws_bedrock_runtime_endpoint
        else:
            base_url = f"https://bedrock-agentcore.{region}.{get_aws_dns_suffix(region)}"

        # Based on boto3 client.invoke_agent_runtime, the path is:
        # /runtimes/{URL-ENCODED-ARN}/invocations?qualifier=<value>
        encoded_arn: Final = quote(agent_runtime_arn, safe="")
        endpoint_url = f"{base_url}/runtimes/{encoded_arn}/invocations"

        # Add qualifier as query parameter if provided
        if "qualifier" in optional_params:
            endpoint_url = f"{endpoint_url}?qualifier={optional_params['qualifier']}"

        return endpoint_url

    def sign_request(
        self,
        headers: dict,
        optional_params: dict,
        request_data: dict,
        api_base: str,
        api_key: str | None = None,
        model: str | None = None,
        stream: bool | None = None,
        fake_stream: bool | None = None,
    ) -> tuple[dict, bytes | None]:
        # Set Accept header required by MCP servers on AgentCore
        # Per MCP spec (Streamable HTTP transport): client MUST include Accept header
        # listing both application/json and text/event-stream as supported content types
        headers["Accept"] = "application/json, text/event-stream"

        # Check if api_key (bearer token) is provided for Cognito authentication
        # Priority: api_key parameter first, then optional_params
        jwt_token: Final = api_key or optional_params.get("api_key")
        if jwt_token:
            verbose_logger.debug(
                "AgentCore: Using Bearer token authentication (Cognito/JWT) - token: %s...", jwt_token[:50]
            )
            headers["Content-Type"] = "application/json"
            headers["Authorization"] = f"Bearer {jwt_token}"
            # Return headers with bearer token and JSON-encoded body (not SigV4 signed)
            return headers, json.dumps(request_data).encode()

        # Otherwise, use AWS SigV4 authentication
        verbose_logger.debug("AgentCore: Using AWS SigV4 authentication (IAM)")
        return self._sign_request(
            service_name="bedrock-agentcore",
            headers=headers,
            optional_params=optional_params,
            request_data=request_data,
            api_base=api_base,
            model=model,
            stream=stream,
            fake_stream=fake_stream,
            api_key=api_key,
        )

    def _get_agent_runtime_arn(self, model: str) -> str:
        """
        Extract ARN from model string
        model = "agentcore/arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/hosted_agent_r9jvp-3ySZuRHjLC"
        returns: "arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/hosted_agent_r9jvp-3ySZuRHjLC"
        """
        parts: Final = model.split("/", 1)
        if len(parts) != 2 or parts[0] != "agentcore":
            raise ValueError(
                "Invalid model format. Expected format: 'model=bedrock/agentcore/arn:aws:bedrock-agentcore:region:account:runtime/runtime_id'"
            )
        return parts[1]

    def _extract_region_from_arn(self, arn: str) -> str:
        """
        Extract region from ARN
        arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/hosted_agent_r9jvp-3ySZuRHjLC
        returns: us-west-2
        """
        parts: Final = arn.split(":")
        if len(parts) >= 4:
            return parts[3]
        raise ValueError(f"Invalid ARN format: {arn}")

    def _get_runtime_session_id(self, optional_params: dict) -> str:
        """
        Get or generate runtime session ID (must be 33+ chars)
        """
        session_id: Final = optional_params.get("runtimeSessionId", None)
        if session_id:
            verbose_logger.debug("Using provided runtimeSessionId: %s", session_id)
            return session_id

        # Generate a session ID with 33+ characters
        generated_id: Final = f"litellm-session-{uuid.uuid4()}"
        verbose_logger.debug("Generated new session ID: %s", generated_id)
        return generated_id

    def _get_runtime_user_id(self, optional_params: dict) -> str | None:
        """
        Get runtime user ID if provided
        """
        user_id: Final = optional_params.get("runtimeUserId", None)
        if user_id:
            verbose_logger.debug("Using provided runtimeUserId: %s", user_id)
        return user_id

    def transform_request(
        self,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        headers: dict,
    ) -> dict:
        """
        Transform the request to AgentCore format.

        Based on boto3's implementation:
        - Session ID goes in header: X-Amzn-Bedrock-AgentCore-Runtime-Session-Id
        - User ID goes in header: X-Amzn-Bedrock-AgentCore-Runtime-User-Id
        - Qualifier goes as query parameter
        - Only the payload goes in the request body

        Payload shape:
        - ``prompt`` is always present and contains the text-only flatten of the
          last message's content (existing behavior).
        - ``content`` is added ONLY when the ``forward_multimodal_content`` litellm
          param is truthy AND the last message's ``content`` is a list containing a
          non-text block (e.g. ``image_url``, ``file``, ``input_audio``). The list is
          forwarded verbatim so the agent's ``@app.entrypoint`` handler can parse the
          OpenAI-shaped multimodal blocks. This is opt-in because an AgentCore agent
          must be explicitly written to read ``payload["content"]``; by default the
          payload stays byte-identical to the legacy ``{"prompt": "..."}`` shape.

        Returns:
            dict: Payload dict containing the prompt and (optionally) the OpenAI
            content list.
        """
        verbose_logger.debug("AgentCore transform_request - optional_params keys: %s", list(optional_params.keys()))

        # Use the last message content as the prompt
        prompt: Final = convert_content_list_to_str(messages[-1])

        # Create the payload - this is what goes in the body (raw JSON)
        payload: Final[dict] = {"prompt": prompt}

        # Opt-in: when forward_multimodal_content is set, forward the OpenAI content
        # list verbatim under "content" so an attachment-aware agent can read the raw
        # blocks (image_url, file, etc.). Default off keeps the payload byte-identical
        # to the legacy {"prompt": "..."} shape for agents that only read the prompt.
        if self._should_forward_multimodal_content(optional_params, litellm_params):
            last_content: Final = messages[-1].get("content")
            if isinstance(last_content, list) and any(
                isinstance(block, dict) and block.get("type") not in (None, "text") for block in last_content
            ):
                # Copy so the payload never aliases messages[-1]["content"]; shallow,
                # not deep, to avoid cloning large base64 media on the request path.
                payload["content"] = list(last_content)

        # Get or generate session ID - this goes in the header
        runtime_session_id: Final = self._get_runtime_session_id(optional_params)
        headers["X-Amzn-Bedrock-AgentCore-Runtime-Session-Id"] = runtime_session_id

        # Get user ID if provided - this goes in the header
        runtime_user_id: Final = self._get_runtime_user_id(optional_params)
        if runtime_user_id:
            headers["X-Amzn-Bedrock-AgentCore-Runtime-User-Id"] = runtime_user_id

        # The request data is the payload dict (will be JSON encoded by the HTTP handler)
        # Qualifier will be handled as a query parameter in get_complete_url

        verbose_logger.debug("PAYLOAD: %s", payload)
        return payload

    @staticmethod
    def _should_forward_multimodal_content(optional_params: dict, litellm_params: dict) -> bool:
        """Whether to forward raw OpenAI content blocks under ``payload["content"]``.

        Opt-in via the ``forward_multimodal_content`` litellm param (default ``False``)
        because AgentCore agents must be explicitly written to read the field. The
        value may arrive as a bool or a config/env string ("true", "1", ...). Checks
        ``optional_params`` first (where other AgentCore params land), then
        ``litellm_params``.
        """
        for source in (optional_params, litellm_params):
            if not isinstance(source, dict):
                continue
            value = source.get("forward_multimodal_content")
            if value is None:
                continue
            if isinstance(value, str):
                return value.strip().lower() in ("1", "true", "yes", "on")
            return bool(value)
        return False

    def _extract_sse_json(self, line: str) -> dict | None:
        """Extract and parse JSON from an SSE data line."""
        if not line.startswith("data:"):
            return None

        json_str: Final = line[5:].strip()
        if not json_str:
            return None

        try:
            data: Final = json.loads(json_str)
            # Skip non-dict data (some lines contain JSON strings)
            return data if isinstance(data, dict) else None
        except json.JSONDecodeError:
            verbose_logger.debug("Skipping non-JSON line: %s", line[:100])
            return None

    def _extract_usage_from_event(self, event_data: dict) -> AgentCoreUsage | None:
        """Extract usage information from event metadata."""
        event_payload: Final = event_data.get("event")
        if not event_payload:
            return None

        metadata: Final = event_payload.get("metadata")
        if metadata and "usage" in metadata:
            return metadata["usage"]

        return None

    def _extract_content_delta(self, event_data: dict) -> str | None:
        """Extract text content from contentBlockDelta event."""
        event_payload: Final = event_data.get("event")
        if not event_payload:
            return None

        content_block_delta: Final = event_payload.get("contentBlockDelta")
        if not content_block_delta:
            return None

        delta: Final = content_block_delta.get("delta", {})
        return delta.get("text")

    def _extract_content_from_message(self, message: AgentCoreMessage) -> str:
        """
        Extract text content from message content blocks.
        This works for both SSE messages and JSON responses.
        """
        content_list: Final = message.get("content", [])
        if not isinstance(content_list, list):
            return ""

        return "".join(block["text"] for block in content_list if isinstance(block, dict) and "text" in block)

    def _calculate_usage(self, model: str, messages: list[AllMessageValues], content: str) -> Usage | None:
        """
        Calculate token usage using LiteLLM's token counter.

        Args:
            model: The model name
            messages: Input messages
            content: Response content

        Returns:
            Usage object with calculated tokens, or None if calculation fails
        """
        try:
            from litellm.utils import token_counter

            prompt_tokens: Final = token_counter(model=model, messages=messages)
            completion_tokens: Final = token_counter(model=model, text=content, count_response_tokens=True)
            total_tokens: Final = prompt_tokens + completion_tokens

            verbose_logger.debug(
                "Calculated usage - prompt: %s, completion: %s, total: %s",
                prompt_tokens,
                completion_tokens,
                total_tokens,
            )

            return Usage(
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                total_tokens=total_tokens,
            )
        except Exception as e:
            verbose_logger.warning("Failed to calculate token usage: %s", e)
            return None

    def _parse_json_response(self, response_json: dict) -> AgentCoreParsedResponse:
        """
        Parse direct JSON response (non-streaming).

        Supports multiple agent response schemas:
        0. {"jsonrpc": "2.0", "result": {"message": {"parts": [...]}}} - A2A JSON-RPC
        1. {"result": {"role": "assistant", "content": [{"text": "..."}]}} - standard AgentCore
        2. {"response": [{"text": "..."}]} - Strands agent format
        3. {"result": "plain text"} or {"response": "plain text"} - simple string
        4. Fallback: raw JSON as content string
        """
        # Guard: if json.loads() returned a non-dict (e.g. array or primitive),
        # skip strategy matching and fall back to raw JSON string
        if not isinstance(response_json, dict):
            verbose_logger.warning("AgentCore: JSON response is not a dict. Returning raw JSON as content.")
            return AgentCoreParsedResponse(
                content=json.dumps(response_json),
                usage=None,
                final_message=None,
            )

        # Strategy 0: A2A JSON-RPC format
        # {"jsonrpc": "2.0", "result": {"message": {"parts": [{"kind": "text", "text": "..."}]}}}
        if "jsonrpc" in response_json:
            content = extract_text_from_a2a_response(response_json)
            if content:
                return AgentCoreParsedResponse(
                    content=content,
                    usage=None,
                    final_message=None,
                )
            # Fall through to other strategies if A2A extraction returned empty

        # Strategy 1: {"result": {"content": [{"text": "..."}]}} - standard AgentCore format
        if "result" in response_json and isinstance(response_json["result"], dict):
            result: Final = response_json["result"]
            content = self._extract_content_from_message(result)
            return AgentCoreParsedResponse(
                content=content,
                usage=None,
                final_message=result,
            )

        # Strategy 2: {"response": [{"text": "..."}]} - Strands agent content blocks
        if "response" in response_json and isinstance(response_json["response"], list):
            content = self._extract_content_from_message({"content": response_json["response"]})
            return AgentCoreParsedResponse(
                content=content,
                usage=None,
                final_message=None,
            )

        # Strategy 3: string values - {"result": "text"} or {"response": "text"}
        for key in ("result", "response"):
            val = response_json.get(key)
            if isinstance(val, str):
                return AgentCoreParsedResponse(
                    content=val,
                    usage=None,
                    final_message=None,
                )

        # Strategy 4: fallback - return raw JSON as content
        verbose_logger.warning(
            "AgentCore: Could not extract content from JSON response keys %s. Returning raw JSON as content.",
            list(response_json.keys()),
        )
        return AgentCoreParsedResponse(
            content=json.dumps(response_json),
            usage=None,
            final_message=None,
        )

    def _get_parsed_response(self, raw_response: httpx.Response) -> AgentCoreParsedResponse:
        """
        Parse AgentCore response based on content type.

        Args:
            raw_response: Raw HTTP response from AgentCore

        Returns:
            AgentCoreParsedResponse: Parsed response data
        """
        content_type: Final = raw_response.headers.get("content-type", "").lower()
        verbose_logger.debug("AgentCore response Content-Type: %s", content_type)

        # Parse response based on content type
        if "application/json" in content_type:
            # Direct JSON response
            verbose_logger.debug("Parsing JSON response")
            response_json: Final = raw_response.json()
            verbose_logger.debug("Response JSON: %s", response_json)
            return self._parse_json_response(response_json)
        else:
            # SSE stream response (text/event-stream or default)
            verbose_logger.debug("Parsing SSE stream response")
            response_text: Final = raw_response.text
            verbose_logger.debug("AgentCore response (first 500 chars): %s", response_text[:500])
            return self._parse_sse_stream(response_text)

    def _parse_sse_stream(self, response_text: str) -> AgentCoreParsedResponse:
        """
        Parse Server-Sent Events (SSE) stream format.
        Each line starts with 'data:' followed by JSON.

        Returns:
            AgentCoreParsedResponse: Parsed response with content, usage, and message
        """
        final_message: AgentCoreMessage | None = None
        usage_data: AgentCoreUsage | None = None
        content_blocks: Final[list[str]] = []

        for line in response_text.strip().split("\n"):
            line = line.strip()
            if not line:
                continue

            data = self._extract_sse_json(line)
            if not data:
                continue

            verbose_logger.debug("SSE event keys: %s", list(data.keys()))

            # Check for final complete message
            if "message" in data and isinstance(data["message"], dict):
                final_message = data["message"]
                verbose_logger.debug("Found final message")

            # Process event data
            if "event" in data and isinstance(data["event"], dict):
                event_payload = data["event"]
                verbose_logger.debug("Event payload keys: %s", list(event_payload.keys()))

                # Extract usage metadata
                if usage := self._extract_usage_from_event(data):
                    usage_data = usage
                    verbose_logger.debug("Found usage data: %s", usage_data)

                # Collect content deltas
                if text := self._extract_content_delta(data):
                    content_blocks.append(text)

        # Build final content
        content: Final = self._extract_content_from_message(final_message) if final_message else "".join(content_blocks)

        verbose_logger.debug("Final usage_data: %s", usage_data)

        return AgentCoreParsedResponse(content=content, usage=usage_data, final_message=final_message)

    def _stream_agentcore_response_sync(
        self,
        response: httpx.Response,
        model: str,
    ):
        """
        Internal sync generator that parses SSE and yields ModelResponse chunks.
        """
        buffer = ""
        for text_chunk in response.iter_text():
            buffer += text_chunk

            # Process complete lines
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                line = line.strip()

                if not line or not line.startswith("data:"):
                    continue

                json_str = line[5:].strip()
                if not json_str:
                    continue

                try:
                    data_obj = json.loads(json_str)
                    if not isinstance(data_obj, dict):
                        continue

                    # Process contentBlockDelta events
                    if "event" in data_obj and isinstance(data_obj["event"], dict):
                        event_payload = data_obj["event"]
                        content_block_delta = event_payload.get("contentBlockDelta")

                        if content_block_delta:
                            delta = content_block_delta.get("delta", {})
                            text = delta.get("text", "")

                            if text:
                                chunk = ModelResponseStream(
                                    id=f"chatcmpl-{uuid.uuid4()}",
                                    created=0,
                                    model=model,
                                    object="chat.completion.chunk",
                                )
                                chunk.choices = [
                                    StreamingChoices(
                                        finish_reason=None,
                                        index=0,
                                        delta=Delta(content=text, role="assistant"),
                                    )
                                ]
                                yield chunk

                        # Process metadata/usage
                        metadata = event_payload.get("metadata")
                        if metadata and "usage" in metadata:
                            chunk = ModelResponseStream(
                                id=f"chatcmpl-{uuid.uuid4()}",
                                created=0,
                                model=model,
                                object="chat.completion.chunk",
                            )
                            chunk.choices = [
                                StreamingChoices(
                                    finish_reason="stop",
                                    index=0,
                                    delta=Delta(),
                                )
                            ]
                            usage_data: AgentCoreUsage = metadata["usage"]
                            setattr(
                                chunk,
                                "usage",
                                Usage(
                                    prompt_tokens=usage_data.get("inputTokens", 0),
                                    completion_tokens=usage_data.get("outputTokens", 0),
                                    total_tokens=usage_data.get("totalTokens", 0),
                                ),
                            )
                            yield chunk

                    # Process final message
                    if "message" in data_obj and isinstance(data_obj["message"], dict):
                        chunk = ModelResponseStream(
                            id=f"chatcmpl-{uuid.uuid4()}",
                            created=0,
                            model=model,
                            object="chat.completion.chunk",
                        )
                        chunk.choices = [
                            StreamingChoices(
                                finish_reason="stop",
                                index=0,
                                delta=Delta(),
                            )
                        ]
                        yield chunk

                except json.JSONDecodeError:
                    verbose_logger.debug("Skipping non-JSON SSE line: %s", line[:100])
                    continue

    def get_sync_custom_stream_wrapper(
        self,
        model: str,
        custom_llm_provider: str,
        logging_obj: LiteLLMLoggingObj,
        api_base: str,
        headers: dict,
        data: dict,
        messages: list,
        client: Union[HTTPHandler, "AsyncHTTPHandler"] | None = None,
        json_mode: bool | None = None,
        signed_json_body: bytes | None = None,
    ) -> "CustomStreamWrapper":
        """
        Simplified sync streaming - returns a generator that yields ModelResponse chunks.
        """
        from litellm.llms.custom_httpx.http_handler import (
            HTTPHandler,
            _get_httpx_client,
        )

        if client is None or not isinstance(client, HTTPHandler):
            client = _get_httpx_client(params={})

        verbose_logger.debug("Making sync streaming request to: %s", api_base)

        # Make streaming request
        response: Final = client.post(
            api_base,
            headers=headers,
            data=signed_json_body if signed_json_body else json.dumps(data),
            stream=True,
            logging_obj=logging_obj,
        )

        if response.status_code != 200:
            raise BedrockError(status_code=response.status_code, message=str(response.read()))

        # LOGGING
        logging_obj.post_call(
            input=messages,
            api_key="",
            original_response="first stream response received",
            additional_args={"complete_input_dict": data},
        )

        # Check if response is JSON (agent used sync return) instead of SSE
        content_type: Final = response.headers.get("content-type", "").lower()
        if "application/json" in content_type:
            verbose_logger.debug(
                "AgentCore streaming: received JSON response instead of SSE, converting to single-chunk stream"
            )
            try:
                body: Final = response.read()
                response_json: Final = json.loads(body)
            except (json.JSONDecodeError, Exception) as e:
                raise BedrockError(
                    status_code=response.status_code,
                    message=f"AgentCore: Failed to read/parse JSON response body: {e}",
                )
            parsed: Final = self._parse_json_response(response_json)

            def _json_as_sync_stream():
                # Content chunk
                content_chunk: Final = ModelResponseStream(
                    id=f"chatcmpl-{uuid.uuid4()}",
                    created=0,
                    model=model,
                    object="chat.completion.chunk",
                )
                content_chunk.choices = [
                    StreamingChoices(
                        finish_reason=None,
                        index=0,
                        delta=Delta(content=parsed["content"], role="assistant"),
                    )
                ]
                yield content_chunk

                # Stop sentinel chunk (matches SSE path convention)
                stop_chunk: Final = ModelResponseStream(
                    id=f"chatcmpl-{uuid.uuid4()}",
                    created=0,
                    model=model,
                    object="chat.completion.chunk",
                )
                stop_chunk.choices = [
                    StreamingChoices(
                        finish_reason="stop",
                        index=0,
                        delta=Delta(),
                    )
                ]
                yield stop_chunk

            return CustomStreamWrapper(
                completion_stream=_json_as_sync_stream(),
                model=model,
                custom_llm_provider="bedrock",
                logging_obj=logging_obj,
            )

        # SSE stream (text/event-stream or default) - use existing SSE parser
        return CustomStreamWrapper(
            completion_stream=self._stream_agentcore_response_sync(response, model),
            model=model,
            custom_llm_provider="bedrock",
            logging_obj=logging_obj,
        )

    async def _stream_agentcore_response(
        self,
        response: httpx.Response,
        model: str,
    ) -> AsyncGenerator[ModelResponseStream, None]:
        """
        Internal async generator that parses SSE and yields ModelResponse chunks.
        """
        buffer = ""
        async for text_chunk in response.aiter_text():
            buffer += text_chunk

            # Process complete lines
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                line = line.strip()

                if not line or not line.startswith("data:"):
                    continue

                json_str = line[5:].strip()
                if not json_str:
                    continue

                try:
                    data_obj = json.loads(json_str)
                    if not isinstance(data_obj, dict):
                        continue

                    # Process contentBlockDelta events
                    if "event" in data_obj and isinstance(data_obj["event"], dict):
                        event_payload = data_obj["event"]
                        content_block_delta = event_payload.get("contentBlockDelta")

                        if content_block_delta:
                            delta = content_block_delta.get("delta", {})
                            text = delta.get("text", "")

                            if text:
                                chunk = ModelResponseStream(
                                    id=f"chatcmpl-{uuid.uuid4()}",
                                    created=0,
                                    model=model,
                                    object="chat.completion.chunk",
                                )
                                chunk.choices = [
                                    StreamingChoices(
                                        finish_reason=None,
                                        index=0,
                                        delta=Delta(content=text, role="assistant"),
                                    )
                                ]
                                yield chunk

                        # Process metadata/usage
                        metadata = event_payload.get("metadata")
                        if metadata and "usage" in metadata:
                            chunk = ModelResponseStream(
                                id=f"chatcmpl-{uuid.uuid4()}",
                                created=0,
                                model=model,
                                object="chat.completion.chunk",
                            )
                            chunk.choices = [
                                StreamingChoices(
                                    finish_reason="stop",
                                    index=0,
                                    delta=Delta(),
                                )
                            ]
                            usage_data: AgentCoreUsage = metadata["usage"]
                            setattr(
                                chunk,
                                "usage",
                                Usage(
                                    prompt_tokens=usage_data.get("inputTokens", 0),
                                    completion_tokens=usage_data.get("outputTokens", 0),
                                    total_tokens=usage_data.get("totalTokens", 0),
                                ),
                            )
                            yield chunk

                    # Process final message
                    if "message" in data_obj and isinstance(data_obj["message"], dict):
                        chunk = ModelResponseStream(
                            id=f"chatcmpl-{uuid.uuid4()}",
                            created=0,
                            model=model,
                            object="chat.completion.chunk",
                        )
                        chunk.choices = [
                            StreamingChoices(
                                finish_reason="stop",
                                index=0,
                                delta=Delta(),
                            )
                        ]
                        yield chunk

                except json.JSONDecodeError:
                    verbose_logger.debug("Skipping non-JSON SSE line: %s", line[:100])
                    continue

    async def get_async_custom_stream_wrapper(
        self,
        model: str,
        custom_llm_provider: str,
        logging_obj: LiteLLMLoggingObj,
        api_base: str,
        headers: dict,
        data: dict,
        messages: list,
        client: Optional["AsyncHTTPHandler"] = None,
        json_mode: bool | None = None,
        signed_json_body: bytes | None = None,
    ) -> "CustomStreamWrapper":
        """
        Simplified async streaming - returns an async generator that yields ModelResponse chunks.
        """
        from litellm.llms.custom_httpx.http_handler import (
            AsyncHTTPHandler,
            get_async_httpx_client,
        )

        if client is None or not isinstance(client, AsyncHTTPHandler):
            client = get_async_httpx_client(llm_provider=cast(Any, "bedrock"), params={})

        verbose_logger.debug("Making async streaming request to: %s", api_base)

        # Make async streaming request
        response: Final = await client.post(
            api_base,
            headers=headers,
            data=signed_json_body if signed_json_body else json.dumps(data),
            stream=True,
            logging_obj=logging_obj,
        )

        if response.status_code != 200:
            raise BedrockError(status_code=response.status_code, message=str(await response.aread()))

        # LOGGING
        logging_obj.post_call(
            input=messages,
            api_key="",
            original_response="first stream response received",
            additional_args={"complete_input_dict": data},
        )

        # Check if response is JSON (agent used sync return) instead of SSE
        content_type: Final = response.headers.get("content-type", "").lower()
        if "application/json" in content_type:
            verbose_logger.debug(
                "AgentCore streaming: received JSON response instead of SSE, converting to single-chunk stream"
            )
            try:
                body: Final = await response.aread()
                response_json: Final = json.loads(body)
            except (json.JSONDecodeError, Exception) as e:
                raise BedrockError(
                    status_code=response.status_code,
                    message=f"AgentCore: Failed to read/parse JSON response body: {e}",
                )
            parsed: Final = self._parse_json_response(response_json)

            async def _json_as_async_stream() -> AsyncGenerator[ModelResponseStream, None]:
                # Content chunk
                content_chunk: Final = ModelResponseStream(
                    id=f"chatcmpl-{uuid.uuid4()}",
                    created=0,
                    model=model,
                    object="chat.completion.chunk",
                )
                content_chunk.choices = [
                    StreamingChoices(
                        finish_reason=None,
                        index=0,
                        delta=Delta(content=parsed["content"], role="assistant"),
                    )
                ]
                yield content_chunk

                # Stop sentinel chunk (matches SSE path convention)
                stop_chunk: Final = ModelResponseStream(
                    id=f"chatcmpl-{uuid.uuid4()}",
                    created=0,
                    model=model,
                    object="chat.completion.chunk",
                )
                stop_chunk.choices = [
                    StreamingChoices(
                        finish_reason="stop",
                        index=0,
                        delta=Delta(),
                    )
                ]
                yield stop_chunk

            return CustomStreamWrapper(
                completion_stream=_json_as_async_stream(),
                model=model,
                custom_llm_provider="bedrock",
                logging_obj=logging_obj,
            )

        # SSE stream (text/event-stream or default) - use existing SSE parser
        return CustomStreamWrapper(
            completion_stream=self._stream_agentcore_response(response, model),
            model=model,
            custom_llm_provider="bedrock",
            logging_obj=logging_obj,
        )

    @property
    def has_custom_stream_wrapper(self) -> bool:
        """Indicates that this config has custom streaming support."""
        return True

    @property
    def supports_stream_param_in_request_body(self) -> bool:
        """
        AgentCore does not allow passing `stream` in the request body.
        Streaming is automatic based on the response format.
        """
        return False

    def transform_response(
        self,
        model: str,
        raw_response: httpx.Response,
        model_response: ModelResponse,
        logging_obj: LiteLLMLoggingObj,
        request_data: dict,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        encoding: "tiktoken.Encoding | None",
        api_key: str | None = None,
        json_mode: bool | None = None,
    ) -> ModelResponse:
        """
        Transform the AgentCore response to LiteLLM ModelResponse format.
        AgentCore can return either JSON or SSE (Server-Sent Events) stream responses.

        Note: For streaming responses, use get_streaming_response() instead.
        """
        try:
            # Parse the response based on content type (JSON or SSE)
            parsed_data: Final = self._get_parsed_response(raw_response)

            content: Final = parsed_data["content"]
            usage_data: Final = parsed_data["usage"]

            verbose_logger.debug("Parsed content length: %s", len(content))
            verbose_logger.debug("Usage data: %s", usage_data)

            # Create the message
            message: Final = Message(content=content, role="assistant")

            # Create choices
            choice: Final = Choices(finish_reason="stop", index=0, message=message)

            # Update model response
            model_response.choices = [choice]
            model_response.model = model

            # Add usage information if available
            # Note: AgentCore JSON responses don't include usage data
            # SSE responses may include usage in metadata events
            if usage_data:
                usage: Final = Usage(
                    prompt_tokens=usage_data.get("inputTokens", 0),
                    completion_tokens=usage_data.get("outputTokens", 0),
                    total_tokens=usage_data.get("totalTokens", 0),
                )
                setattr(model_response, "usage", usage)
            else:
                # Calculate token usage using LiteLLM's token counter
                verbose_logger.debug("No usage data from AgentCore - calculating tokens")
                calculated_usage: Final = self._calculate_usage(model, messages, content)
                if calculated_usage:
                    setattr(model_response, "usage", calculated_usage)

            return model_response

        except Exception as e:
            verbose_logger.error("Error processing Bedrock AgentCore response: %s", e)
            raise BedrockError(
                message=f"Error processing response: {e}",
                status_code=raw_response.status_code,
            )

    def validate_environment(
        self,
        headers: dict,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        api_key: str | None = None,
        api_base: str | None = None,
    ) -> dict:
        return headers

    def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
        return BedrockError(status_code=status_code, message=error_message)

    def should_fake_stream(
        self,
        model: str | None,
        stream: bool | None,
        custom_llm_provider: str | None = None,
    ) -> bool:
        # AgentCore supports true streaming - don't buffer
        return False
