o
    ưi                     @   s   d Z ddlmZmZmZ ddlmZ ddlmZm	Z	 ddl
mZmZmZmZ ddlmZ er4ddlmZ dZz
ddlmZ d	ZW n eyK   d
ZY nw G dd dZ	
	
	
ddedee dee dee def
ddZ	ddededddeddf
ddZd
S )z`
A2A Protocol Exception Mapping Utils.

Maps A2A SDK exceptions to LiteLLM A2A exception types.
    )TYPE_CHECKINGAnyOptional)verbose_logger)fix_agent_card_urlis_localhost_or_internal_url)A2AAgentCardErrorA2AConnectionErrorA2AErrorA2ALocalhostURLError)CONNECTION_ERROR_PATTERNS)	A2AClientFTNc                   @   sV   e Zd ZdZededefddZedee defddZ	ededefd	d
Z
dS )A2AExceptionCheckerszA
    Helper class for checking various A2A error conditions.
    	error_strreturnc                    s,   t | tsdS |   t fddtD S )z
        Check if an error string indicates a connection error.

        Args:
            error_str: The error string to check

        Returns:
            True if the error indicates a connection issue
        Fc                 3       | ]}| v V  qd S N .0patternZerror_str_lowerr   c/home/app/Keep/.python/lib/python3.10/site-packages/litellm/a2a_protocol/exception_mapping_utils.py	<genexpr>8       z;A2AExceptionCheckers.is_connection_error.<locals>.<genexpr>)
isinstancestrloweranyr   )r   r   r   r   is_connection_error)   s   
z(A2AExceptionCheckers.is_connection_errorurlc                 C   s   t | S )z
        Check if a URL is a localhost/internal URL.

        Args:
            url: The URL to check

        Returns:
            True if the URL is localhost/internal
        )r   )r    r   r   r   is_localhost_url:   s   z%A2AExceptionCheckers.is_localhost_urlc                    s4   t | tsdS |   g d}t fdd|D S )z
        Check if an error string indicates an agent card error.

        Args:
            error_str: The error string to check

        Returns:
            True if the error is related to agent card fetching/parsing
        F)z
agent cardz
agent-cardz.well-knownzcard not foundzinvalid agentc                 3   r   r   r   r   r   r   r   r   ]   r   z;A2AExceptionCheckers.is_agent_card_error.<locals>.<genexpr>)r   r   r   r   )r   Zagent_card_patternsr   r   r   is_agent_card_errorG   s
   
z(A2AExceptionCheckers.is_agent_card_errorN)__name__
__module____qualname____doc__staticmethodr   boolr   r   r!   r"   r   r   r   r   r   $   s    r   original_exceptioncard_urlapi_basemodelr   c                 C   st   t | }|r|rt|rt|rt||| |dt|r&t|||dt|r4t||p0||dt||d)as  
    Map an A2A SDK exception to a LiteLLM A2A exception type.

    Args:
        original_exception: The original exception from the A2A SDK
        card_url: The URL from the agent card (if available)
        api_base: The original API base URL
        model: The model/agent name

    Returns:
        A mapped LiteLLM A2A exception

    Raises:
        A2ALocalhostURLError: If the error is a connection error to a localhost URL
        A2AConnectionError: If the error is a general connection error
        A2AAgentCardError: If the error is related to agent card issues
        A2AError: For other A2A-related errors
    )localhost_urlbase_urloriginal_errorr,   )messager    r,   )r0   r,   )	r   r   r!   r   r   r"   r   r	   r
   )r)   r*   r+   r,   r   r   r   r   map_a2a_exception`   s>   

r1   error
agent_card
a2a_clientA2AClientTypeis_streamingc                 C   sf   t rtdu r
td|rdnd}td| d| j d| j d| j d		 t|| j t|j	j
|d
S )a  
    Handle A2ALocalhostURLError by fixing the URL and creating a new client.

    This is called when we catch an A2ALocalhostURLError and want to retry
    with the corrected URL.

    Args:
        error: The localhost URL error
        agent_card: The agent card object to fix
        a2a_client: The current A2A client
        is_streaming: Whether this is a streaming request (for logging)

    Returns:
        A new A2A client with the fixed URL

    Raises:
        ImportError: If the A2A SDK is not installed
    NzRA2A SDK is required for localhost retry handling. Install it with: pip install a2az
streaming  zA2A zrequest to 'z
' failed: zF. Agent card contains localhost/internal URL. Retrying with base_url 'z'.)httpx_clientr3   )A2A_SDK_AVAILABLE
_A2AClientImportErrorr   warningr-   r/   r.   r   
_transportr8   )r2   r3   r4   r6   Zrequest_typer   r   r   handle_a2a_localhost_retry   s   r>   )NNN)F)r&   typingr   r   r   Zlitellm._loggingr   Z"litellm.a2a_protocol.card_resolverr   r   Zlitellm.a2a_protocol.exceptionsr   r	   r
   r   Zlitellm.constantsr   Z
a2a.clientr   r5   r9   r:   r;   r   	Exceptionr   r1   r(   r>   r   r   r   r   <module>   sT    >
C