server.py

"""academic-intel-mcp — academic research intelligence for autonomous agents.

Part of the FoundryNet Data Network. Papers from OpenAlex (primary) + arXiv +
PubMed, embedded with fastembed for pgvector similar-paper search. 7 tools + free
mint_info. Free tier 25/day, then x402 (USDC on Solana). Daily ingest ~4am PT.
Transport: Streamable HTTP at /mcp (+ /sse). Health: /health.
"""
from __future__ import annotations

import asyncio
import contextlib
import inspect
import logging
import time

from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import event_log

import academic_aggregator as agg
import config
import core
import daily_curator
import identity
import payment_gate
import x402_standard
import supa
import tools

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("acad.mcp")

if not supa.configured():
    logger.warning("SUPABASE_SERVICE_KEY not set — dataset disabled until configured.")

mcp = FastMCP("academic-intel")

if payment_gate.is_active():
    logger.info(f"pay-per-query ARMED → {config.PAYMENT_RECIPIENT} after {config.FREE_TIER_DAILY}/day free")
else:
    logger.info("pay-per-query INERT — all tools free")

tools.register_all(mcp)


# ── okf-reliability-v1: emit reliability metadata on every tool result (#2964) ──
try:
    from okf_middleware import ReliabilityMiddleware
    mcp.add_middleware(ReliabilityMiddleware(server_id="academic-intel"))
except Exception as _okf_e:  # noqa: BLE001
    import logging as _okf_log; _okf_log.getLogger(__name__).warning(f"okf middleware not wired: {_okf_e}")


@mcp.custom_route("/v1/reliability", methods=["GET"])
async def _okf_reliability_route(request):
    from starlette.responses import JSONResponse
    import okf_endpoint
    return JSONResponse(okf_endpoint.reliability_payload("academic-intel"))


@mcp.custom_route("/health", methods=["GET"])
async def health(request: Request) -> JSONResponse:
    return JSONResponse({
        "status": "ok", "service": "academic-intel-mcp", "transport": "streamable-http",
        "network": "FoundryNet Data Network",
        "tools": ["search_papers", "paper_detail", "citation_graph", "author_profile",
                  "trending_research", "similar_papers", "daily_brief", "mint_info"],
        "dataset": "supabase:papers" if supa.configured() else "unconfigured",
        "sources": "openalex + arxiv + pubmed (keyless)" + (" + semantic_scholar" if config.S2_API_KEY else ""),
        "embeddings": config.EMBED_MODEL,
        "x402_enabled": config.X402_ENABLED,
        "query_payment": "armed" if payment_gate.is_active() else "free",
        "free_tier_daily": config.FREE_TIER_DAILY,
        "payment_recipient": config.PAYMENT_RECIPIENT,
    })


@mcp.custom_route("/ping", methods=["GET"])
async def ping(request: Request) -> JSONResponse:
    return JSONResponse({"status": "ok"})


# ── REST surface ─────────────────────────────────────────────────────────────
_ERR = {"bad_request": 400, "not_configured": 503, "not_found": 404,
        "payment_required": 402, "embedding_unavailable": 503}


def _resp(d: dict) -> JSONResponse:
    if "error" not in d:
        return JSONResponse(d, status_code=200)
    err = str(d.get("error") or "")
    code = _ERR.get(err, 502 if err in ("network", "non_json_response", "unreachable") else 400)
    if err.startswith("http_") and err[5:].isdigit():
        code = int(err[5:])
    return JSONResponse(d, status_code=code)


async def _body(request: Request) -> dict:
    try:
        b = await request.json()
        return b if isinstance(b, dict) else {}
    except Exception:
        return {}


def _akey(request: Request, body: dict) -> str:
    return identity.resolve_agent_key(body.get("agent_id"), request=request)


@mcp.custom_route("/v1/search", methods=["POST"])
async def rest_search(request: Request) -> JSONResponse:
    b = await _body(request)
    filters = {k: b.get(k) for k in ("query", "year_from", "year_to", "field", "venue",
                                     "min_citations", "open_access_only", "limit")}
    return _resp(await core.do_search(filters, agent_key=_akey(request, b),
                                      payment_tx=b.get("payment_tx"), api_key=identity.bearer(request)))


@mcp.custom_route("/v1/paper", methods=["POST"])
async def rest_paper(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_detail(b.get("paper_id"), b.get("doi"), b.get("title")))


@mcp.custom_route("/v1/citations", methods=["POST"])
async def rest_citations(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_citation_graph(b.get("paper_id", ""), b.get("direction", "citations"),
                                              b.get("depth", 1), agent_key=_akey(request, b),
                                              payment_tx=b.get("payment_tx"), api_key=identity.bearer(request)))


@mcp.custom_route("/v1/author", methods=["POST"])
async def rest_author(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_author(b.get("author_name"), b.get("author_id"),
                                      agent_key=_akey(request, b), payment_tx=b.get("payment_tx"),
                                      api_key=identity.bearer(request)))


@mcp.custom_route("/v1/trending", methods=["POST"])
async def rest_trending(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_trending(b.get("field"), b.get("days", 30), b.get("metric", "citations"),
                                        agent_key=_akey(request, b), payment_tx=b.get("payment_tx"),
                                        api_key=identity.bearer(request)))


@mcp.custom_route("/v1/similar", methods=["POST"])
async def rest_similar(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_similar(b.get("query"), b.get("paper_id"),
                                       agent_key=_akey(request, b), payment_tx=b.get("payment_tx"),
                                       api_key=identity.bearer(request)))


@mcp.custom_route("/v1/brief", methods=["POST"])
async def rest_brief(request: Request) -> JSONResponse:
    b = await _body(request)
    return _resp(await core.do_daily_brief(b.get("date"), agent_key=_akey(request, b),
                                           payment_tx=b.get("payment_tx"),
                                           api_key=identity.bearer(request),
                                           stripe_token=b.get("stripe_token") or identity.stripe_token(request)))


@mcp.custom_route("/v1/mint-info", methods=["GET", "POST"])
async def rest_mint(request: Request) -> JSONResponse:
    return JSONResponse(core.mint_info())


@mcp.custom_route("/admin/aggregate", methods=["POST"])
async def admin_aggregate(request: Request) -> JSONResponse:
    import os
    tok = os.environ.get("ADMIN_TOKEN", "")
    if not tok or request.headers.get("x-admin-token") != tok:
        return JSONResponse({"error": "forbidden"}, status_code=403)
    qp = request.query_params
    hours = int(qp["hours"]) if qp.get("hours", "").isdigit() else None
    if qp.get("wait") == "1":
        return JSONResponse(await agg.run_aggregation(hours))
    asyncio.create_task(agg.run_aggregation(hours))
    return JSONResponse({"started": True, "hours": hours})


# ── Discovery ────────────────────────────────────────────────────────────────
_TAGLINE = "Academic paper search, citations, authors & semantic related-work for agents."
_DESC = ("Academic research intelligence for agents: academic paper search, scientific literature, "
         "citation analysis, arXiv search, author metrics, trending research, and semantic related-work "
         "discovery (pgvector). Part of the FoundryNet Data Network — attest research with MINT "
         "Protocol; see also gov-contracts, brand-intel, patent-intel, financial-signals, weather-intel, "
         "compliance, cyber-intel.")
_KEYWORDS = ["academic papers", "research search", "scientific literature", "citation analysis",
             "arXiv search", "paper search", "literature review"]

_TOOL_NAMES = ["search_papers", "paper_detail", "citation_graph", "author_profile",
               "trending_research", "similar_papers", "daily_brief", "mint_info"]

_AGENT_CARD = {
    "name": "Academic Research Intelligence MCP",
    "description": ("Search academic papers, citations, and authors, and find related work with "
                    "semantic similarity — across OpenAlex, arXiv, and PubMed — for literature reviews."),
    "url": config.PUBLIC_MCP_URL,
    "version": "1.0.0",
    "capabilities": {"tools": _TOOL_NAMES},
    "provider": {"name": "FoundryNet", "url": "https://foundrynet.io"},
    "network": "FoundryNet Data Network",
    "attestation": {"protocol": "MINT Protocol", "endpoint": config.MINT_MCP_URL,
                    "verified_outputs": True, "live_feed": "https://mint.foundrynet.io/feed", "feed_api": "https://mint-mcp-production.up.railway.app/v1/feed"},
    "protocols": {"mcp": {"endpoint": config.PUBLIC_MCP_URL, "transport": "streamable-http",
                          "tools_count": len(_TOOL_NAMES)},
                  "x402": {"supported": True, "currency": "USDC", "network": "solana"}},
    "see_also": config.SISTER_SERVERS, "mint_protocol": config.MINT_MCP_URL,
    "contact": "hello@foundrynet.io",
}


@mcp.custom_route("/.well-known/agent-card.json", methods=["GET"])
async def agent_card(request: Request) -> JSONResponse:
    return JSONResponse(_AGENT_CARD, headers={"Cache-Control": "public, max-age=300"})


@mcp.custom_route("/.well-known/mcp", methods=["GET"])
async def mcp_endpoints(request: Request) -> JSONResponse:
    return JSONResponse({"endpoints": [{"url": config.PUBLIC_MCP_URL, "transport": "streamable-http",
                                        "name": "Academic Research Intelligence MCP"}]},
                        headers={"Cache-Control": "public, max-age=300"})


async def _live_tools() -> list:
    res = mcp.list_tools()
    if inspect.iscoroutine(res):
        res = await res
    return [{"name": t.name, "description": (getattr(t, "description", "") or "").strip(),
             "inputSchema": getattr(t, "parameters", None) or {"type": "object"}} for t in res]


@mcp.custom_route("/.well-known/mcp/server-card.json", methods=["GET"])
async def server_card(request: Request) -> JSONResponse:
    live = await _live_tools()
    return JSONResponse({
        "serverInfo": {"name": "Academic Research Intelligence MCP", "version": "1.0.0"},
        "authentication": {"type": "http", "scheme": "bearer",
                           "description": ("paper_detail and mint_info are free; other tools give 25 free "
                                           "queries/day then take an fnet_ Bearer key OR x402 USDC.")},
        "tools": live, "version": "1.0", "name": "Academic Research Intelligence MCP",
        "tagline": _TAGLINE, "description": _DESC,
        "serverUrl": config.PUBLIC_MCP_URL, "transport": "streamable-http",
        "tools_count": len(live),
        "categories": ["research", "academic", "data", "science", "literature"],
        "keywords": _KEYWORDS, "network": "FoundryNet Data Network",
        "see_also": config.SISTER_SERVERS,
        "pricing": {"model": "metered",
                    "free_tier": f"{config.FREE_TIER_DAILY} queries/day + free paper_detail",
                    "paid_from": f"{config.PRICE_SEARCH} USDC per query (x402)"},
    }, headers={"Cache-Control": "public, max-age=300"})


# ── Daily aggregation (~4am PT = AGG_HOUR_UTC) ───────────────────────────────
async def _agg_loop():
    while True:
        now = time.gmtime()
        secs = now.tm_hour * 3600 + now.tm_min * 60 + now.tm_sec
        wait = (config.AGG_HOUR_UTC * 3600 - secs) % 86400 or 86400
        try:
            await asyncio.sleep(wait)
            if supa.configured():
                await agg.run_aggregation()
        except asyncio.CancelledError:
            break
        except Exception as e:  # noqa: BLE001
            logger.warning(f"agg loop: {e}")
            await asyncio.sleep(3600)


_FREE_TOOL_NAMES = {"mint_info", "detail"}


@mcp.custom_route("/.well-known/mcp.json", methods=["GET"])
async def wellknown_mcp_json(request: Request) -> JSONResponse:
    """Machine-discovery card (emerging standard) for AI clients/crawlers."""
    live = await _live_tools()
    names = [t["name"] for t in live]
    return JSONResponse({
        "name": _AGENT_CARD["name"],
        "description": _AGENT_CARD["description"],
        "url": config.PUBLIC_MCP_URL,
        "transport": ["streamable-http"],
        "tools": names,
        "pricing": {"model": "per-query", "free_tier": True,
                    "paid_tools": [n for n in names if n not in _FREE_TOOL_NAMES]},
        "attestation": {"enabled": True, "protocol": "MINT Protocol",
                        "feed": "https://mint.foundrynet.io/feed"},
        "network": {"name": "FoundryNet Data Network", "servers": 17,
                    "homepage": "https://foundrynet.io"},
    }, headers={"Cache-Control": "public, max-age=300"})



# ── Standard x402 compliance (discoverable on x402scan / 402 Index / CDP Bazaar) ──
@mcp.custom_route("/x402", methods=["GET"])
async def x402_index(request: Request) -> JSONResponse:
    return JSONResponse(x402_standard.index(),
                        headers={"Cache-Control": "public, max-age=300",
                                 "Access-Control-Allow-Origin": "*"})


@mcp.custom_route("/.well-known/x402", methods=["GET"])
async def x402_wellknown(request: Request) -> JSONResponse:
    return JSONResponse(x402_standard.index(),
                        headers={"Cache-Control": "public, max-age=300",
                                 "Access-Control-Allow-Origin": "*"})


@mcp.custom_route("/x402/{tool}", methods=["GET", "POST"])
async def x402_resource(request: Request) -> JSONResponse:
    tool = request.path_params["tool"]
    if tool not in x402_standard.PAID_TOOLS:
        return JSONResponse({"error": "unknown_resource", "tool": tool,
                             "available": list(x402_standard.PAID_TOOLS)}, status_code=404)
    challenge = x402_standard.payment_required_header(tool)
    return JSONResponse(x402_standard.payment_required(tool), status_code=402,
                        headers={"Cache-Control": "public, max-age=300",
                                 "Access-Control-Allow-Origin": "*",
                                 "PAYMENT-REQUIRED": challenge,
                                 "X-PAYMENT": challenge,
                                 "Link": '</openapi.json>; rel="describedby"',
                                 "WWW-Authenticate": 'x402 version="2"'})


@mcp.custom_route("/openapi.json", methods=["GET"])
async def openapi_doc(request: Request) -> JSONResponse:
    """OpenAPI 3.1 discovery doc — x402scan requires a spec at a discoverable URL."""
    return JSONResponse(x402_standard.openapi(),
                        headers={"Cache-Control": "public, max-age=300",
                                 "Access-Control-Allow-Origin": "*",
                                 "Link": '</openapi.json>; rel="describedby"'})


def build_dual_app():
    main_app = mcp.http_app(transport="http", path="/mcp")
    sse_app = mcp.http_app(transport="sse", path="/sse")
    for r in sse_app.routes:
        if getattr(r, "path", None) in ("/sse", "/messages"):
            main_app.router.routes.append(r)
    main_life, sse_life = main_app.router.lifespan_context, sse_app.router.lifespan_context

    @contextlib.asynccontextmanager
    async def _dual_lifespan(app):
        async with main_life(app):
            async with sse_life(app):
                task = asyncio.create_task(_agg_loop())
                brief_task = asyncio.create_task(daily_curator.run_daily_loop())
                try:
                    yield
                finally:
                    for t in (task, brief_task):
                        t.cancel()
                        with contextlib.suppress(Exception):
                            await t
    main_app.router.lifespan_context = _dual_lifespan
    # Per-call telemetry middleware (fire-and-forget to agents ingest).
    main_app.add_middleware(BaseHTTPMiddleware, dispatch=event_log.middleware)
    return main_app


if __name__ == "__main__":
    import uvicorn
    logger.info(f"academic-intel-mcp starting on 0.0.0.0:{config.PORT} "
                f"(dataset={'supabase' if supa.configured() else 'off'}, x402={config.X402_ENABLED})")
    uvicorn.run(build_dual_app(), host="0.0.0.0", port=config.PORT, log_level="warning")