"""Tools adapter for converting MCP tools to LangChain tools.

This module provides functionality to convert MCP tools into LangChain-compatible
tools, handle tool execution, and manage tool conversion between the two formats.
"""

from collections.abc import Awaitable, Callable
from typing import Annotated, Any, TypedDict, get_args

from langchain_core.messages import ToolMessage
from langchain_core.messages.content import (
    FileContentBlock,
    ImageContentBlock,
    TextContentBlock,
    create_file_block,
    create_image_block,
    create_text_block,
)
from langchain_core.tools import (
    BaseTool,
    InjectedToolArg,
    StructuredTool,
    ToolException,
)
from langchain_core.tools.base import get_all_basemodel_annotations
from mcp import ClientSession
from mcp.server.fastmcp.tools import Tool as FastMCPTool
from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata
from mcp.types import (
    AudioContent,
    BlobResourceContents,
    ContentBlock,
    EmbeddedResource,
    ImageContent,
    ResourceLink,
    TextContent,
    TextResourceContents,
)
from mcp.types import Tool as MCPTool
from pydantic import BaseModel, create_model

from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks, _MCPCallbacks
from langchain_mcp_adapters.interceptors import (
    MCPToolCallRequest,
    MCPToolCallResult,
    ToolCallInterceptor,
)
from langchain_mcp_adapters.sessions import Connection, create_session

try:
    # langgraph installed
    from langgraph.types import Command

    LANGGRAPH_PRESENT = True
except ImportError:
    LANGGRAPH_PRESENT = False

# Type alias for LangChain content blocks used in ToolMessage
ToolMessageContentBlock = TextContentBlock | ImageContentBlock | FileContentBlock

# Conditional type based on langgraph availability
if LANGGRAPH_PRESENT:
    ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage | Command
else:
    ConvertedToolResult = list[ToolMessageContentBlock] | ToolMessage

MAX_ITERATIONS = 1000


def _summarize_tool_error(tool_content: list[ToolMessageContentBlock]) -> str:
    """Build a human-readable error message from converted error content blocks.

    Joins the text from all text blocks. Non-text blocks (image/file) carry no
    human-readable error text, so fall back to a summary rather than dumping
    their raw repr (which may include large base64 payloads). An error with no
    content at all gets a minimal placeholder.

    Args:
        tool_content: The converted LangChain content blocks from an MCP
            `isError=True` result.

    Returns:
        The joined text from all text blocks, a summary if there are only
            non-text blocks, or a placeholder if there is no content.
    """
    error_parts: list[str] = [
        block["text"] for block in tool_content if block["type"] == "text"
    ]
    if error_parts:
        return "\n".join(error_parts)
    if tool_content:
        return (
            "MCP tool returned an error with no text content "
            f"({len(tool_content)} non-text content block(s))."
        )
    return "MCP tool returned an error with empty content."


class _MCPToolExecutionError(ToolException):
    """MCP tool execution error (`CallToolResult(isError=True)`).

    Carries the already-converted LangChain content blocks so the error can be
    surfaced to the model as a failed tool output instead of crashing the run.
    The exception message is derived from `tool_content` (see
    `_summarize_tool_error`) when the error is constructed; the message is what
    the model sees when error handling is disabled and the exception is raised.
    The message is a snapshot taken at construction and is *not* recomputed, so
    treat `tool_content` as read-only afterward — mutating it would leave the
    message out of sync with the content.

    Deliberately narrow: only `isError=True` results use this type.
    Transport/session failures and content-conversion errors are *not*
    `ToolException` subclasses, so they bypass `handle_tool_error` and
    propagate (see `_handle_mcp_tool_error`).
    """

    def __init__(self, tool_content: list[ToolMessageContentBlock]) -> None:
        super().__init__(_summarize_tool_error(tool_content))
        self.tool_content = tool_content


def _handle_mcp_tool_error(
    error: ToolException,
) -> list[ToolMessageContentBlock]:
    """Surface MCP tool execution errors as failed tool output.

    Used as the `handle_tool_error` callback on the generated
    `StructuredTool`. LangChain only routes `ToolException` (and subclasses)
    to this callback, so it never sees other error types: content-conversion
    failures (`NotImplementedError`, `ValueError`) and transport/session
    failures are not `ToolException` subclasses and propagate without ever
    reaching here.

    Args:
        error: The `ToolException` raised during tool execution.

    Returns:
        The content blocks carried by an `_MCPToolExecutionError` (an MCP
            `isError=True` result), so the caller produces a `ToolMessage` with
            `status="error"`. Real content (text or image/file) is preserved
            verbatim. Only when the MCP error carried no content at all
            (`content=[]`) is a single minimal text block substituted, so the
            `ToolMessage` is not empty (a fragile shape for some model providers);
            that placeholder is an adapter-level fallback, not server-provided
            error detail.

    Raises:
        ToolException: Re-raised for any non-`_MCPToolExecutionError`
            `ToolException` so it still propagates. Defensive: the adapter only
            ever raises `_MCPToolExecutionError` into this path, but an
            interceptor raising a bare `ToolException` must keep propagating
            rather than be swallowed.
    """
    if isinstance(error, _MCPToolExecutionError):
        if error.tool_content:
            return error.tool_content
        return [create_text_block(text=str(error))]
    raise error


class MCPToolArtifact(TypedDict):
    """Artifact returned from MCP tool calls.

    This TypedDict wraps the structured content from MCP tool calls,
    allowing for future extension if MCP adds more fields to tool results.

    Attributes:
        structured_content: The structured content returned by the MCP tool,
            corresponding to the structuredContent field in CallToolResult.
    """

    structured_content: dict[str, Any]


def _convert_mcp_content_to_lc_block(  # noqa: PLR0911
    content: ContentBlock,
) -> ToolMessageContentBlock:
    """Convert any MCP content block to a LangChain content block.

    Args:
        content: MCP content object (TextContent, ImageContent, AudioContent,
            ResourceLink, or EmbeddedResource).

    Returns:
        LangChain content block dict.

    Raises:
        NotImplementedError: If AudioContent is passed.
        ValueError: If an unknown content type is passed.
    """
    if isinstance(content, TextContent):
        return create_text_block(text=content.text)

    if isinstance(content, ImageContent):
        return create_image_block(base64=content.data, mime_type=content.mimeType)

    if isinstance(content, AudioContent):
        msg = (
            "AudioContent conversion to LangChain content blocks is not yet "
            f"supported. Received audio with mime type: {content.mimeType}"
        )
        raise NotImplementedError(msg)

    if isinstance(content, ResourceLink):
        mime_type = content.mimeType or None
        if mime_type and mime_type.startswith("image/"):
            return create_image_block(url=str(content.uri), mime_type=mime_type)
        return create_file_block(url=str(content.uri), mime_type=mime_type)

    if isinstance(content, EmbeddedResource):
        resource = content.resource
        if isinstance(resource, TextResourceContents):
            return create_text_block(text=resource.text)
        if isinstance(resource, BlobResourceContents):
            mime_type = resource.mimeType or None
            if mime_type and mime_type.startswith("image/"):
                return create_image_block(base64=resource.blob, mime_type=mime_type)
            return create_file_block(base64=resource.blob, mime_type=mime_type)
        msg = f"Unknown embedded resource type: {type(resource).__name__}"
        raise ValueError(msg)

    msg = f"Unknown MCP content type: {type(content).__name__}"
    raise ValueError(msg)


def _convert_call_tool_result(
    call_tool_result: MCPToolCallResult,
) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
    """Convert MCP MCPToolCallResult to LangChain tool result format.

    Converts MCP content blocks to LangChain content blocks:
    - TextContent -> {"type": "text", "text": ...}
    - ImageContent -> {"type": "image", "base64": ..., "mime_type": ...}
    - ResourceLink (image/*) -> {"type": "image", "url": ..., "mime_type": ...}
    - ResourceLink (other) -> {"type": "file", "url": ..., "mime_type": ...}
    - EmbeddedResource (text) -> {"type": "text", "text": ...}
    - EmbeddedResource (blob) -> {"type": "image", ...} or {"type": "file", ...}
    - AudioContent -> raises NotImplementedError

    Args:
        call_tool_result: The result from calling an MCP tool. Can be either
            a `CallToolResult` (MCP format), a `ToolMessage` (LangChain format),
            or a `Command` (LangGraph format, if `langgraph` is installed).

    Returns:
        A tuple containing:
        - The content: either a string (single text), list of content blocks,
            `ToolMessage`, or `Command`
        - The artifact: `MCPToolArtifact` with `structured_content` if present,
            otherwise None

    Raises:
        _MCPToolExecutionError: If the MCP tool reported an execution error
            (`CallToolResult(isError=True)`).
        NotImplementedError: If AudioContent is encountered.
        ValueError: If an unknown or unsupported content type is encountered.
    """
    # If the interceptor returned a ToolMessage directly, return it as the content
    # with None as the artifact to match the content_and_artifact format
    if isinstance(call_tool_result, ToolMessage):
        return call_tool_result, None

    # If the interceptor returned a Command (LangGraph), return it directly
    if LANGGRAPH_PRESENT and isinstance(call_tool_result, Command):
        return call_tool_result, None

    # Convert all MCP content blocks to LangChain content blocks
    tool_content: list[ToolMessageContentBlock] = [
        _convert_mcp_content_to_lc_block(content)
        for content in call_tool_result.content
    ]

    if call_tool_result.isError:
        raise _MCPToolExecutionError(tool_content)

    # Extract structured content and wrap in MCPToolArtifact
    artifact: MCPToolArtifact | None = None
    if call_tool_result.structuredContent is not None:
        artifact = MCPToolArtifact(
            structured_content=call_tool_result.structuredContent
        )

    return tool_content, artifact


def _build_interceptor_chain(
    base_handler: Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]],
    tool_interceptors: list[ToolCallInterceptor] | None,
) -> Callable[[MCPToolCallRequest], Awaitable[MCPToolCallResult]]:
    """Build composed handler chain with interceptors in onion pattern.

    Args:
        base_handler: Innermost handler executing the actual tool call.
        tool_interceptors: Optional list of interceptors to wrap the handler.

    Returns:
        Composed handler with all interceptors applied. First interceptor
        in list becomes outermost layer.
    """
    handler = base_handler

    if tool_interceptors:
        for interceptor in reversed(tool_interceptors):
            current_handler = handler

            async def wrapped_handler(
                req: MCPToolCallRequest,
                _interceptor: ToolCallInterceptor = interceptor,
                _handler: Callable[
                    [MCPToolCallRequest], Awaitable[MCPToolCallResult]
                ] = current_handler,
            ) -> MCPToolCallResult:
                return await _interceptor(req, _handler)

            handler = wrapped_handler

    return handler


async def _list_all_tools(session: ClientSession) -> list[MCPTool]:
    """List all available tools from an MCP session with pagination support.

    Args:
        session: The MCP client session.

    Returns:
        A list of all available MCP tools.

    Raises:
        RuntimeError: If maximum iterations exceeded while listing tools.
    """
    current_cursor: str | None = None
    all_tools: list[MCPTool] = []

    iterations = 0

    while True:
        iterations += 1
        if iterations > MAX_ITERATIONS:
            msg = "Reached max of 1000 iterations while listing tools."
            raise RuntimeError(msg)

        list_tools_page_result = await session.list_tools(cursor=current_cursor)

        if list_tools_page_result.tools:
            all_tools.extend(list_tools_page_result.tools)

        # Pagination spec: https://modelcontextprotocol.io/specification/2025-06-18/server/utilities/pagination
        # compatible with None or ""
        if not list_tools_page_result.nextCursor:
            break

        current_cursor = list_tools_page_result.nextCursor
    return all_tools


def convert_mcp_tool_to_langchain_tool(
    session: ClientSession | None,
    tool: MCPTool,
    *,
    connection: Connection | None = None,
    callbacks: Callbacks | None = None,
    tool_interceptors: list[ToolCallInterceptor] | None = None,
    server_name: str | None = None,
    tool_name_prefix: bool = False,
    handle_tool_errors: bool = True,
) -> BaseTool:
    """Convert an MCP tool to a LangChain tool.

    NOTE: this tool can be executed only in a context of an active MCP client session.

    Args:
        session: MCP client session
        tool: MCP tool to convert
        connection: Optional connection config to use to create a new session
                    if a `session` is not provided
        callbacks: Optional callbacks for handling notifications and events
        tool_interceptors: Optional list of interceptors for tool call processing
        server_name: Name of the server this tool belongs to
        tool_name_prefix: If `True` and `server_name` is provided, the tool name will be
            prefixed w/ server name (e.g., `"weather_search"` instead of `"search"`)
        handle_tool_errors: If `True` (default), an MCP tool execution error
            (`CallToolResult(isError=True)`) is returned to the model as a
            `ToolMessage` with `status="error"` so the agent can self-correct
            instead of crashing the run. If `False`, a `ToolException` is raised
            instead (legacy behavior). Transport/session failures and
            content-conversion errors (e.g. unsupported audio content) always
            raise regardless of this setting; only MCP execution errors
            (`isError=True`) are governed by it.

    Returns:
        a LangChain tool

    """
    if session is None and connection is None:
        msg = "Either a session or a connection config must be provided"
        raise ValueError(msg)

    async def call_tool(
        runtime: Annotated[object | None, InjectedToolArg()] = None,
        **arguments: dict[str, Any],
    ) -> tuple[ConvertedToolResult, MCPToolArtifact | None]:
        """Execute tool call with interceptor chain and return formatted result.

        Args:
            runtime: LangGraph tool runtime if available, otherwise None.
            **arguments: Tool arguments as keyword args.

        Returns:
            A tuple of (content, artifact) where:
            - content: string, list of strings/content blocks, ToolMessage, or Command
            - artifact: MCPToolArtifact with structured_content if present, else None
        """
        mcp_callbacks = (
            callbacks.to_mcp_format(
                context=CallbackContext(server_name=server_name, tool_name=tool.name)
            )
            if callbacks is not None
            else _MCPCallbacks()
        )

        # Create the innermost handler that actually executes the tool call
        async def execute_tool(request: MCPToolCallRequest) -> MCPToolCallResult:
            """Execute the actual MCP tool call with optional session creation.

            Args:
                request: Tool call request with name, args, headers, and context.

            Returns:
                MCPToolCallResult from MCP SDK.

            Raises:
                ValueError: If neither session nor connection provided.
                RuntimeError: If tool call returns None.
            """
            tool_name = request.name
            tool_args = request.args
            effective_connection = connection

            # If headers were modified, create a new connection with updated headers
            modified_headers = request.headers
            if modified_headers is not None and connection is not None:
                # Create a new connection config with updated headers
                updated_connection = dict(connection)
                if connection["transport"] in (
                    "sse",
                    "http",
                    "streamable_http",
                    "streamable-http",
                ):
                    existing_headers = connection.get("headers", {})
                    updated_connection["headers"] = {
                        **existing_headers,
                        **modified_headers,
                    }
                    effective_connection = updated_connection

            captured_exception = None

            if session is None:
                # If a session is not provided, we will create one on the fly
                if effective_connection is None:
                    msg = "Either session or connection must be provided"
                    raise ValueError(msg)

                async with create_session(
                    effective_connection, mcp_callbacks=mcp_callbacks
                ) as tool_session:
                    await tool_session.initialize()
                    try:
                        call_tool_result = await tool_session.call_tool(
                            tool_name,
                            tool_args,
                            progress_callback=mcp_callbacks.progress_callback,
                        )
                    except Exception as e:  # noqa: BLE001
                        # Capture exception to re-raise outside context manager
                        captured_exception = e

                # Re-raise the exception outside the context manager
                # This is necessary because the context manager may suppress exceptions
                # This change was introduced to work-around an issue in MCP SDK
                # that may suppress exceptions when the client disconnects.
                # If this is causing an issue, with your use case, please file an issue
                # on the langchain-mcp-adapters GitHub repo.
                if captured_exception is not None:
                    raise captured_exception
            else:
                call_tool_result = await session.call_tool(
                    tool_name,
                    tool_args,
                    progress_callback=mcp_callbacks.progress_callback,
                )

            return call_tool_result

        # Build and execute the interceptor chain
        handler = _build_interceptor_chain(execute_tool, tool_interceptors)
        request = MCPToolCallRequest(
            name=tool.name,
            args=arguments,
            server_name=server_name or "unknown",
            headers=None,
            runtime=runtime,
        )
        call_tool_result = await handler(request)

        return _convert_call_tool_result(call_tool_result)

    meta = getattr(tool, "meta", None)
    base = tool.annotations.model_dump() if tool.annotations is not None else {}
    meta = {"_meta": meta} if meta is not None else {}
    metadata = {**base, **meta} or None

    # Apply server name prefix if requested
    lc_tool_name = tool.name
    if tool_name_prefix and server_name:
        lc_tool_name = f"{server_name}_{tool.name}"

    # `handle_tool_error` is typed to return `str`, but our callback returns a
    # list of LangChain content blocks. langchain_core's output formatting
    # preserves list content (it is recognized message content), so the blocks
    # reach the `ToolMessage` unchanged. The type mismatch is intentional and
    # load-bearing; see `_handle_mcp_tool_error`. This relies on langchain_core's
    # current content-block recognition and is locked by
    # `test_mcp_tool_error_preserves_non_text_content`.
    error_handler = _handle_mcp_tool_error if handle_tool_errors else False
    return StructuredTool(
        name=lc_tool_name,
        description=tool.description or "",
        args_schema=tool.inputSchema,
        coroutine=call_tool,
        response_format="content_and_artifact",
        metadata=metadata,
        handle_tool_error=error_handler,  # type: ignore[arg-type]
    )


async def load_mcp_tools(
    session: ClientSession | None,
    *,
    connection: Connection | None = None,
    callbacks: Callbacks | None = None,
    tool_interceptors: list[ToolCallInterceptor] | None = None,
    server_name: str | None = None,
    tool_name_prefix: bool = False,
    handle_tool_errors: bool = True,
) -> list[BaseTool]:
    """Load all available MCP tools and convert them to LangChain [tools](https://docs.langchain.com/oss/python/langchain/tools).

    Args:
        session: The MCP client session. If `None`, connection must be provided.
        connection: Connection config to create a new session if session is `None`.
        callbacks: Optional `Callbacks` for handling notifications and events.
        tool_interceptors: Optional list of interceptors for tool call processing.
        server_name: Name of the server these tools belong to.
        tool_name_prefix: If `True` and `server_name` is provided, tool names will be
            prefixed w/ server name (e.g., `"weather_search"` instead of `"search"`).
        handle_tool_errors: If `True` (default), an MCP tool execution error
            (`CallToolResult(isError=True)`) is returned to the model as a
            `ToolMessage` with `status="error"` so the agent can self-correct
            instead of crashing the run. If `False`, a `ToolException` is raised
            instead (legacy behavior). Transport/session failures and
            content-conversion errors (e.g. unsupported audio content) always
            raise regardless of this setting; only MCP execution errors
            (`isError=True`) are governed by it.

    Returns:
        List of LangChain [tools](https://docs.langchain.com/oss/python/langchain/tools).
            Tool annotations are returned as part of the tool metadata object.

    Raises:
        ValueError: If neither session nor connection is provided.
    """
    if session is None and connection is None:
        msg = "Either a session or a connection config must be provided"
        raise ValueError(msg)

    mcp_callbacks = (
        callbacks.to_mcp_format(context=CallbackContext(server_name=server_name))
        if callbacks is not None
        else _MCPCallbacks()
    )

    if session is None:
        # If a session is not provided, we will create one on the fly
        if connection is None:
            msg = "Either session or connection must be provided"
            raise ValueError(msg)
        async with create_session(
            connection, mcp_callbacks=mcp_callbacks
        ) as tool_session:
            await tool_session.initialize()
            tools = await _list_all_tools(tool_session)
    else:
        tools = await _list_all_tools(session)

    return [
        convert_mcp_tool_to_langchain_tool(
            session,
            tool,
            connection=connection,
            callbacks=callbacks,
            tool_interceptors=tool_interceptors,
            server_name=server_name,
            tool_name_prefix=tool_name_prefix,
            handle_tool_errors=handle_tool_errors,
        )
        for tool in tools
    ]


def _get_injected_args(tool: BaseTool) -> list[str]:
    """Extract field names with InjectedToolArg annotation from tool schema.

    Args:
        tool: LangChain tool to inspect.

    Returns:
        List of field names marked as injected arguments.
    """

    def _is_injected_arg_type(type_: type) -> bool:
        """Check if type annotation contains InjectedToolArg."""
        return any(
            isinstance(arg, InjectedToolArg)
            or (isinstance(arg, type) and issubclass(arg, InjectedToolArg))
            for arg in get_args(type_)[1:]
        )

    return [
        field
        for field, field_info in get_all_basemodel_annotations(tool.args_schema).items()
        if _is_injected_arg_type(field_info)
    ]


def to_fastmcp(tool: BaseTool) -> FastMCPTool:
    """Convert LangChain tool to FastMCP tool.

    Args:
        tool: LangChain tool to convert.

    Returns:
        FastMCP tool equivalent.

    Raises:
        TypeError: If args_schema is not BaseModel subclass.
        NotImplementedError: If tool has injected arguments.
    """
    if not issubclass(tool.args_schema, BaseModel):
        msg = (
            "Tool args_schema must be a subclass of pydantic.BaseModel. "
            "Tools with dict args schema are not supported."
        )
        raise TypeError(msg)

    parameters = tool.tool_call_schema.model_json_schema()
    field_definitions = {
        field: (field_info.annotation, field_info)
        for field, field_info in tool.tool_call_schema.model_fields.items()
    }
    arg_model = create_model(
        f"{tool.name}Arguments", **field_definitions, __base__=ArgModelBase
    )
    fn_metadata = FuncMetadata(arg_model=arg_model)

    # We'll use an Any type for the function return type.
    # We're providing the parameters separately
    async def fn(**arguments: dict[str, Any]) -> Any:  # noqa: ANN401
        return await tool.ainvoke(arguments)

    injected_args = _get_injected_args(tool)
    if len(injected_args) > 0:
        msg = "LangChain tools with injected arguments are not supported"
        raise NotImplementedError(msg)

    return FastMCPTool(
        fn=fn,
        name=tool.name,
        description=tool.description,
        parameters=parameters,
        fn_metadata=fn_metadata,
        is_async=True,
    )
