Spaces:
Running
Running
File size: 4,744 Bytes
33516f7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """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(),
)
|