Merge pull request #123 from Madhan5422i/fix/revamp-docs-mcp

feat(docs-mcp)

Author: nishant49-code

juspay_docs_mcp/server.py

@@ -14,16 +14,11 @@
 Discovery runs once at module import; results are persisted to snapshot.json.
 """
 
-import json
 import logging
-import os
-import re
-from contextvars import ContextVar
 from typing import Annotated, Optional
-from urllib.parse import urljoin, urlparse
+from urllib.parse import urlparse
 
 import httpx
-from markdownify import markdownify
 from mcp.server.fastmcp import FastMCP
 from pydantic import Field
 
@@ -33,43 +28,25 @@
 logger = logging.getLogger(__name__)
 
 
-# ----------------------------------------------------------------------------
-# Per-request credentials (ContextVar populated by middleware in juspay_mcp.main)
-# ----------------------------------------------------------------------------
-
-juspay_request_credentials: ContextVar[Optional[dict]] = ContextVar(
-    "juspay_request_credentials", default=None
-)
-
+# ---------------------------------------------------------------------------
+# Static domain allowlist
+# ---------------------------------------------------------------------------
 
-def set_juspay_request_credentials(credentials):
-    """Set Juspay credentials for the current request context."""
-    juspay_request_credentials.set(credentials)
+_ALLOWED_SUFFIXES = (".juspay.io", ".juspay.in")
+_ALLOWED_EXACT = frozenset({"juspay.io", "juspay.in", "dth95m2xtyv8v.cloudfront.net"})
+_ALLOWED_SCHEMES = frozenset({"http", "https"})
+_ALLOWED_DOMAINS_DISPLAY = "*.juspay.io, *.juspay.in, dth95m2xtyv8v.cloudfront.net"
 
 
-def get_juspay_request_credentials():
-    """Get Juspay credentials from current request context."""
-    return juspay_request_credentials.get()
-
-
-# ----------------------------------------------------------------------------
-# Transcripts (curated commentary appended to specific URLs in doc_fetch_tool)
-# ----------------------------------------------------------------------------
-
-def _load_transcripts() -> dict:
-    path = os.path.join(os.path.dirname(__file__), "transcripts.json")
-    try:
-        with open(path, "r") as f:
-            return {str(k): str(v) for k, v in json.load(f).items()}
-    except FileNotFoundError:
-        logger.info("No transcripts.json at %s; running without transcripts", path)
-        return {}
-    except json.JSONDecodeError as e:
-        logger.warning("Failed to parse transcripts.json: %s", e)
-        return {}
-
-
-_TRANSCRIPTS_MAP = _load_transcripts()
+def _url_allowed(url: str) -> bool:
+    parsed = urlparse(url)
+    if parsed.scheme not in _ALLOWED_SCHEMES:
+        return False
+    # hostname is lowercased and strips the port.
+    host = parsed.hostname or ""
+    if host in _ALLOWED_EXACT:
+        return True
+    return any(host.endswith(s) for s in _ALLOWED_SUFFIXES)
 
 
 # ----------------------------------------------------------------------------
@@ -112,10 +89,10 @@ def _slug_for(entry: dict) -> Optional[str]:
 _CAT_HINT: str = ", ".join(_CATEGORIES) if _CATEGORIES else "(none discovered)"
 
 logger.info(
-    "Docs MCP initialized: %d products across %d categories, %d allowed domains",
+    "Docs MCP initialized: %d products across %d categories (allowed: %s)",
     len(_ENRICHED_SOURCES),
     len(_CATEGORIES),
-    len(_DOMAINS),
+    _ALLOWED_DOMAINS_DISPLAY,
 )
 
 
@@ -134,39 +111,15 @@ def _slug_for(entry: dict) -> Optional[str]:
 # Helpers
 # ----------------------------------------------------------------------------
 
-_META_REFRESH_RE = re.compile(
-    r'<meta http-equiv="refresh" content="[^;]+;\s*url=([^"]+)"',
-    re.IGNORECASE,
-)
-
 
-def _url_allowed(url: str) -> bool:
-    return any(url.startswith(d) for d in _DOMAINS)
 
 
-async def _fetch_with_processing(url: str) -> str:
-    """Fetch URL, follow meta-refresh once, inject transcript, markdownify."""
+async def _fetch(url: str) -> str:
+    """Fetch URL and return the raw response text."""
     try:
         response = await _HTTPX.get(url)
         response.raise_for_status()
-        content = response.text
-
-        m = _META_REFRESH_RE.search(content)
-        if m:
-            new_url = urljoin(str(response.url), m.group(1))
-            if not _url_allowed(new_url):
-                return (
-                    f"Error: redirect URL not allowed. "
-                    f"Allowed domains: {', '.join(sorted(_DOMAINS))}"
-                )
-            response = await _HTTPX.get(new_url)
-            response.raise_for_status()
-            content = response.text
-
-        if url in _TRANSCRIPTS_MAP:
-            content = content + "\n\n---\n\n" + _TRANSCRIPTS_MAP[url]
-
-        return markdownify(content)
+        return response.text
     except (httpx.HTTPStatusError, httpx.RequestError) as e:
         return f"Encountered an HTTP error: {e}"
 
@@ -255,11 +208,10 @@ async def explore_product(
 ) -> str:
     """Fetch the llms.txt index for a specific Juspay product.
 
-    Looks up the product slug in the discovered catalog and returns its
-    documentation index as markdown. The index contains .md content links
-    that you can read via doc_fetch_tool(url).
+    Looks up the product slug in the catalog and returns the raw llms.txt
+    content. The index contains .md content links readable via doc_fetch_tool(url).
 
-    Returns an error if the slug isn't in the catalog - call list_products()
+    Returns an error if the slug isn't in the catalog — call list_products()
     to see valid slugs.
     """
     entry = _SLUG_INDEX.get(product)
@@ -274,7 +226,7 @@ async def explore_product(
             f"Call list_products() to see all slugs. "
             f"Examples: {sample}{more}."
         )
-    return await _fetch_with_processing(entry["llms_txt"])
+    return await _fetch(entry["llms_txt"])
 
 
 @mcp.tool()
@@ -283,25 +235,24 @@ async def doc_fetch_tool(
         str,
         Field(
             description=(
-                "URL to fetch. Must be on an allowed domain (auto-derived "
-                "from the discovered product catalog, typically juspay.io)."
+                f"URL to fetch. Must be on an allowed domain: {_ALLOWED_DOMAINS_DISPLAY}."
             ),
         ),
     ],
 ) -> str:
-    """Fetch any allowed Juspay docs URL as markdown.
+    """Fetch any allowed Juspay docs URL and return its raw text content.
 
     Use this after explore_product() to read specific pages by URL.
-    Follows meta-refresh redirects once. Returns markdown-converted
-    content. Returns an error if the URL is on a disallowed domain.
+    Returns markdown mirror of the URL.
+    Returns an error if the URL is on a disallowed domain.
     """
     url = url.strip()
     if not _url_allowed(url):
         return (
-            f"Error: URL not allowed. Must start with one of: "
-            f"{', '.join(sorted(_DOMAINS))}"
+            f"Error: URL not on an allowed domain. "
+            f"Allowed: {_ALLOWED_DOMAINS_DISPLAY}"
         )
-    return await _fetch_with_processing(url)
+    return await _fetch(url)
 
 
 # Export the underlying low-level Server for main.py / stdio.py

juspay_docs_mcp/snapshot.json

@@ -1,5 +1,5 @@
 {
-  "fetched_at": "2026-05-15T10:02:42.687046Z",
+  "fetched_at": "2026-05-19T13:47:55.391152Z",
   "sources": [
     {
       "name": "Name: Juspay Billing\n\nEnd to end Mandate lifecycle. By using Billing, you treat Mandates as a configuration, not a codebase",

juspay_mcp/auth/middleware.py

@@ -56,6 +56,7 @@ def __init__(
         cfg: OAuthConfig,
         portal: PortalClient,
         validation_cache: dict[str, tuple[PortalUserInfo, float]] | None = None,
+        skip_path_prefixes: tuple[str, ...] = (),
     ) -> None:
         super().__init__(app)
         self._cfg = cfg
@@ -66,6 +67,7 @@ def __init__(
         self._cache: dict[str, tuple[PortalUserInfo, float]] = (
             validation_cache if validation_cache is not None else {}
         )
+        self._skip_path_prefixes = skip_path_prefixes
 
     async def _validate_with_cache(self, token: str) -> PortalUserInfo | None:
         # Dev bypass: useful for curl-driven smoke tests so we don't need a
@@ -131,7 +133,12 @@ async def dispatch(
         request: Request,
         call_next: Callable[[Request], Awaitable[Response]],
     ) -> Response:
-        if _is_public_path(request.url.path):
+        path = request.url.path
+        if _is_public_path(path):
+            return await call_next(request)
+        if self._skip_path_prefixes and any(
+            path == p or path.startswith(p + "/") for p in self._skip_path_prefixes
+        ):
             return await call_next(request)
 
         auth_header = request.headers.get("authorization") or request.headers.get("Authorization")

juspay_mcp/auth/routes.py

@@ -348,16 +348,18 @@ async def revoke(request: Request) -> Response:
         return JSONResponse({"revoked": revoked})
 
     # ---------- assemble ------------------------------------------------------
+    _WELL_KNOWN_METHODS = ["GET", "OPTIONS"]
+
     routes: list[Route] = [
-        Route("/.well-known/oauth-protected-resource", endpoint=prm_root, methods=["GET"]),
+        Route("/.well-known/oauth-protected-resource", endpoint=prm_root, methods=_WELL_KNOWN_METHODS),
         Route(
             "/{mount:str}/.well-known/oauth-protected-resource",
             endpoint=prm_for_mount,
-            methods=["GET"],
+            methods=_WELL_KNOWN_METHODS,
         ),
-        Route("/.well-known/oauth-authorization-server", endpoint=asm, methods=["GET"]),
-        Route("/.well-known/openid-configuration", endpoint=asm, methods=["GET"]),
-        Route("/.well-known/jwks.json", endpoint=jwks, methods=["GET"]),
+        Route("/.well-known/oauth-authorization-server", endpoint=asm, methods=_WELL_KNOWN_METHODS),
+        Route("/.well-known/openid-configuration", endpoint=asm, methods=_WELL_KNOWN_METHODS),
+        Route("/.well-known/jwks.json", endpoint=jwks, methods=_WELL_KNOWN_METHODS),
         Route("/oauth/register", endpoint=register, methods=["POST"]),
         Route("/oauth/authorize", endpoint=authorize, methods=["GET"]),
         Route("/oauth/callback", endpoint=callback, methods=["GET"]),