butler/mcp/client_http.py

Rama main · réplica de lectura del repositorio original.

"""HTTP / SSE MCP client transport."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

from butler.mcp.config import validate_http_url
from butler.mcp.types import McpServerConfig

logger = logging.getLogger(__name__)


async def connect_http(
    config: McpServerConfig,
    *,
    workspace: Path | None = None,
) -> tuple[Any, list[Any], Any]:
    del workspace  # reserved for future workspace-scoped headers
    err = validate_http_url(config)
    if err:
        raise ValueError(err)

    from mcp import ClientSession

    headers = dict(config.headers or {})
    url = config.url
    transport = None
    read_stream = None
    write_stream = None

    if config.sse:
        from mcp.client.sse import sse_client

        transport = sse_client(url, headers=headers)
        read_stream, write_stream = await transport.__aenter__()
    else:
        from mcp.client.streamable_http import streamablehttp_client

        transport = streamablehttp_client(url, headers=headers)
        opened = await transport.__aenter__()
        if isinstance(opened, tuple) and len(opened) >= 2:
            read_stream, write_stream = opened[0], opened[1]
        else:
            read_stream, write_stream = opened, None

    if write_stream is None:
        raise RuntimeError("HTTP MCP transport did not provide write stream")

    session = ClientSession(read_stream, write_stream)
    await session.__aenter__()
    await session.initialize()
    listed = await session.list_tools()
    tools = list(getattr(listed, "tools", None) or [])

    async def cleanup() -> None:
        from butler.mcp.client_http_ops import (
            close_http_session_safe,
            close_http_transport_safe,
        )

        await close_http_session_safe(session)
        if transport is not None:
            await close_http_transport_safe(transport)

    return session, tools, cleanup


async def call_http_tool(session: Any, tool_name: str, arguments: dict[str, Any]) -> str:
    from butler.mcp.client_stdio import json_dumps_result

    async def _call_with_retry() -> str:
        result = await session.call_tool(tool_name, arguments)
        parts: list[str] = []
        for content in getattr(result, "content", None) or []:
            text = getattr(content, "text", None)
            parts.append(str(text) if text is not None else str(content))
        joined: str = "\n".join(parts) if parts else json_dumps_result(result)
        return joined

    from butler.core.effects import async_with_retry
    from typing import cast

    decorated = async_with_retry(max_attempts=2, wait_seconds=0.1, retry_on=(OSError, ConnectionError))(_call_with_retry)
    result: str = cast(str, await decorated())
    return result