adapter.py

"""TranscriptAPI adapter exposing YouTube tools to Praxist peers.

The plugin gives a research agent four bounded read surfaces over YouTube:
a transcript, a video search, video metadata, and a channel's uploads.

Audience
--------
The tools are consumed by peers at agent-runtime time. Every handler returns
a JSON-serializable ``dict`` with a ``schema_version`` field and never raises.
Errors come back as ``{"error": "..."}`` so the agent can recover and carry on
instead of failing the run.

Configuration
-------------
``TRANSCRIPTAPI_KEY`` env var, sent as ``Authorization: Bearer <key>``. Get one
at https://transcriptapi.com/dashboard/api-keys. When unset, every handler
returns ``{"error": "TRANSCRIPTAPI_KEY is not set"}`` rather than attempting the
request. Praxist treats tool credentials as optional enhancers: a missing key
must degrade only this lookup path and must not break a task that does not use
these tools.

Rate limiting
-------------
The API returns 429 when the plan's budget is exceeded. We retry once after a
short sleep; a second 429 returns an error dict so the model can fall back
rather than stall.
"""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

try:  # claude_agent_sdk is only required when the MCP server is spun up.
    from claude_agent_sdk import (  # type: ignore[import-not-found]
        create_sdk_mcp_server,
        tool,
    )
except ImportError:  # pragma: no cover - SDK missing in some environments
    tool = None
    create_sdk_mcp_server = None


_BASE_URL = os.environ.get("TRANSCRIPTAPI_BASE_URL", "https://transcriptapi.com")
"""Override only for testing against a staging host."""

_DEFAULT_TIMEOUT_SECONDS = 20.0
_RATE_LIMIT_RETRY_DELAY_SECONDS = 1.5
_SCHEMA_VERSION = 1

_MAX_TRANSCRIPT_CHARS = 20000
"""Praxist bounds inline tool output. A long transcript is truncated here and the
response says so, rather than flooding a peer's context window."""


__all__ = [
    "create_tool_plugin",
    "create_transcriptapi_server",
    "youtube_channel_videos",
    "youtube_search",
    "youtube_transcript",
    "youtube_video_info",
]


def _api_key() -> str | None:
    key = os.environ.get("TRANSCRIPTAPI_KEY")
    return key.strip() if key and key.strip() else None


def _request(path: str, params: dict[str, Any]) -> dict[str, Any]:
    """GET one endpoint and normalize every failure into an error dict."""
    key = _api_key()
    if not key:
        return {"schema_version": _SCHEMA_VERSION, "error": "TRANSCRIPTAPI_KEY is not set"}

    query = urllib.parse.urlencode({k: v for k, v in params.items() if v not in (None, "")})
    url = f"{_BASE_URL}{path}?{query}"

    for attempt in range(2):
        req = urllib.request.Request(
            url,
            headers={
                "Authorization": f"Bearer {key}",
                "Accept": "application/json",
                "User-Agent": "praxist-transcriptapi-plugin/0.1.0",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=_DEFAULT_TIMEOUT_SECONDS) as res:
                return json.loads(res.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            if exc.code == 429 and attempt == 0:
                time.sleep(_RATE_LIMIT_RETRY_DELAY_SECONDS)
                continue
            if exc.code == 429:
                return {
                    "schema_version": _SCHEMA_VERSION,
                    "error": "transcriptapi rate-limited after retry",
                    "status": 429,
                }
            if exc.code in (401, 403):
                return {
                    "schema_version": _SCHEMA_VERSION,
                    "error": "transcriptapi rejected the API key",
                    "status": exc.code,
                }
            return {
                "schema_version": _SCHEMA_VERSION,
                "error": f"transcriptapi returned HTTP {exc.code}",
                "status": exc.code,
            }
        except urllib.error.URLError as exc:
            return {"schema_version": _SCHEMA_VERSION, "error": f"transcriptapi unreachable: {exc.reason}"}
        except (TimeoutError, json.JSONDecodeError) as exc:
            return {"schema_version": _SCHEMA_VERSION, "error": f"transcriptapi bad response: {exc}"}

    return {"schema_version": _SCHEMA_VERSION, "error": "transcriptapi request failed"}


# --------------------------------------------------------------------------- #
# Pure handlers (callable from tests without claude_agent_sdk)
# --------------------------------------------------------------------------- #


def youtube_transcript(video_url: str, language: str | None = None) -> dict[str, Any]:
    """Fetch a video transcript. ``video_url`` accepts a URL or an 11-char video ID."""
    raw = _request("/api/v2/youtube/transcript", {"video_url": video_url, "language": language})
    if "error" in raw:
        return raw
    text = raw.get("transcript") or raw.get("text") or ""
    truncated = False
    if isinstance(text, str) and len(text) > _MAX_TRANSCRIPT_CHARS:
        text = text[:_MAX_TRANSCRIPT_CHARS]
        truncated = True
    return {
        "schema_version": _SCHEMA_VERSION,
        "video_url": video_url,
        "language": raw.get("language"),
        "transcript": text,
        "truncated": truncated,
    }


def youtube_search(query: str, kind: str = "video") -> dict[str, Any]:
    """Search YouTube. ``kind`` is ``video`` or ``channel``."""
    raw = _request("/api/v2/youtube/search", {"q": query, "type": kind})
    if "error" in raw:
        return raw
    return {"schema_version": _SCHEMA_VERSION, "query": query, "results": raw.get("results", raw)}


def youtube_video_info(video_url: str) -> dict[str, Any]:
    """Fetch title, channel, duration, and other metadata for one video."""
    raw = _request("/api/v2/youtube/info", {"video_url": video_url})
    if "error" in raw:
        return raw
    return {"schema_version": _SCHEMA_VERSION, "video_url": video_url, "info": raw}


def youtube_channel_videos(channel: str) -> dict[str, Any]:
    """List a channel's latest uploads via RSS. ``channel`` accepts a handle, URL, or UC id."""
    raw = _request("/api/v2/youtube/channel/latest", {"channel": channel})
    if "error" in raw:
        return raw
    return {"schema_version": _SCHEMA_VERSION, "channel": channel, "videos": raw.get("videos", raw)}


# --------------------------------------------------------------------------- #
# MCP server construction
# --------------------------------------------------------------------------- #


def create_transcriptapi_server() -> Any:
    """Build the in-process MCP server Praxist connects peers to."""
    if create_sdk_mcp_server is None or tool is None:
        raise RuntimeError("claude_agent_sdk is required to start the transcriptapi tool server")

    @tool("youtube_transcript", "Get the transcript of a YouTube video.", {"video_url": str})
    async def _transcript(args: dict[str, Any]) -> dict[str, Any]:
        result = youtube_transcript(args["video_url"], args.get("language"))
        return {"content": [{"type": "text", "text": json.dumps(result)}]}

    @tool("youtube_search", "Search YouTube for videos or channels matching a query.", {"query": str})
    async def _search(args: dict[str, Any]) -> dict[str, Any]:
        result = youtube_search(args["query"], args.get("kind", "video"))
        return {"content": [{"type": "text", "text": json.dumps(result)}]}

    @tool("youtube_video_info", "Get metadata for one YouTube video.", {"video_url": str})
    async def _info(args: dict[str, Any]) -> dict[str, Any]:
        result = youtube_video_info(args["video_url"])
        return {"content": [{"type": "text", "text": json.dumps(result)}]}

    @tool("youtube_channel_videos", "List recent uploads for a YouTube channel.", {"channel": str})
    async def _channel(args: dict[str, Any]) -> dict[str, Any]:
        result = youtube_channel_videos(args["channel"])
        return {"content": [{"type": "text", "text": json.dumps(result)}]}

    return create_sdk_mcp_server(
        name="transcriptapi",
        version="0.1.0",
        tools=[_transcript, _search, _info, _channel],
    )


def create_tool_plugin() -> dict[str, object]:
    """Manifest entrypoint exposing the transcriptapi tool server descriptor."""
    return {
        "tool_server_ref": "tool_server:transcriptapi",
        "server_name": "transcriptapi",
        "factory": "adapter:create_transcriptapi_server",
        "tool_names": [
            "youtube_transcript",
            "youtube_search",
            "youtube_video_info",
            "youtube_channel_videos",
        ],
        "visibility": ["peer", "panel"],
        "required_capability": "tool_server.transcriptapi",
        "handlers": {
            "youtube_transcript": "adapter:youtube_transcript",
            "youtube_search": "adapter:youtube_search",
            "youtube_video_info": "adapter:youtube_video_info",
            "youtube_channel_videos": "adapter:youtube_channel_videos",
        },
    }