src/mcp_server/server.py

"""FastMCP 4 Streamable HTTP server: public tools `search` and `fetch`.

Product transport is HTTP (`stateless_http=True`), endpoint `/mcp`.
Does not expose Neo4j / Cypher. Agent and UI LangChain tools stay on the CLIs.
"""

from __future__ import annotations

import os

from fastmcp import FastMCP
from mcp.types import ToolAnnotations
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import JSONResponse

from src.deep_research_agent.tools import SearchResults
from src.mcp_server.caps import (
    DEFAULT_MAX_SEGMENTS,
    DEFAULT_MAX_SUMMARIES,
    DEFAULT_SNIPPET_LIMIT,
)
from src.mcp_server.rate_limit import IPRateLimitMiddleware
from src.mcp_server.retrieval import fetch as fetch_corpus
from src.mcp_server.retrieval import search as search_corpus

mcp = FastMCP(
    name="learningfocused",
    instructions=(
        "Future of Education knowledge base (podcast, Substack, unique YouTube). "
        "Use search for semantic retrieval and fetch for a known episode_id, "
        "doc_id, or YouTube video_id. Podcast episodes have no public transcript site; "
        "fetch(episode_id) returns every transcript segment in time order. "
        "Substack and YouTube hits include canonical_url for the original page. "
        "Default snippets are 2560 characters (~25% of a typical article). "
        "Pass snippet_limit=0 to return the full stored segment or article "
        "(capped at 32768 characters). Truncated hits include omitted_chars. "
        "No graph/Cypher tools."
    ),
)


@mcp.tool(
    name="search",
    annotations=ToolAnnotations(
        title="Search knowledge base",
        readOnlyHint=True,
        idempotentHint=True,
        openWorldHint=False,
    ),
)
def search(
    query: str,
    max_segments: int = DEFAULT_MAX_SEGMENTS,
    max_summaries: int = DEFAULT_MAX_SUMMARIES,
    snippet_limit: int = DEFAULT_SNIPPET_LIMIT,
) -> SearchResults:
    """Search podcast transcripts, Substack articles, and unique YouTube videos.

    Args:
        query: Natural-language search query.
        max_segments: Max transcript/article segments (clamped).
        max_summaries: Max summaries (clamped).
        snippet_limit: Max characters per snippet. Default 2560. Pass 0 for the
            full stored chunk (segment or article), up to 32768 characters.
            Truncated results include omitted_chars and a retry hint.
    """
    return search_corpus(
        query,
        max_segments=max_segments,
        max_summaries=max_summaries,
        snippet_limit=snippet_limit,
    )


@mcp.tool(
    name="fetch",
    annotations=ToolAnnotations(
        title="Fetch source by id",
        readOnlyHint=True,
        idempotentHint=True,
        openWorldHint=False,
    ),
)
def fetch(
    id: str,
    max_segments: int = DEFAULT_MAX_SEGMENTS,
    max_summaries: int = DEFAULT_MAX_SUMMARIES,
    snippet_limit: int = DEFAULT_SNIPPET_LIMIT,
) -> SearchResults:
    """Fetch a known source. Podcast episodes return the full transcript.

    There is no public Art19 transcript page. fetch(episode_id) returns every
    transcript_segment in time order (not a similarity search). Pass
    snippet_limit=0 for the full text of each segment. Substack articles and
    YouTube videos include canonical_url pointing at the original page.

    Args:
        id: Episode id (e.g. S2E335), Substack doc_id, or YouTube video_id.
        max_segments: Unused for episode fetch (all segments are returned, up
            to 40). Still used for unknown ids that fall back to search.
        max_summaries: Max summaries (clamped).
        snippet_limit: Max characters per snippet. Default 2560. Pass 0 for the
            full stored chunk (segment or article), up to 32768 characters.
            Truncated results include omitted_chars and a retry hint.
    """
    return fetch_corpus(
        id,
        max_segments=max_segments,
        max_summaries=max_summaries,
        snippet_limit=snippet_limit,
    )


@mcp.custom_route("/health", methods=["GET"])
async def health_check(_request: Request) -> JSONResponse:
    return JSONResponse({"status": "ok", "service": "learningfocused-mcp"})


def http_middleware() -> list[Middleware]:
    return [Middleware(IPRateLimitMiddleware)]


def build_http_app():
    return mcp.http_app(
        path="/mcp",
        stateless_http=True,
        middleware=http_middleware(),
    )


def run() -> None:
    host = os.getenv("MCP_HOST", "127.0.0.1")
    port = int(os.getenv("PORT") or os.getenv("MCP_PORT") or "8000")
    mcp.run(
        transport="http",
        host=host,
        port=port,
        path="/mcp",
        stateless_http=True,
        middleware=http_middleware(),
    )