from __future__ import annotations
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, ConfigDict, Field, field_validator
from .client import (
InspireHEPClient,
build_author_query,
build_fulltext_query,
build_title_query,
)
mcp = FastMCP("inspirehep_mcp")
class BaseSearchInput(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
large_collaboration: bool = Field(
default=False,
description="Include large-collaboration papers. Default false excludes very large collaborations.",
)
limit: int = Field(
default=20,
ge=1,
le=50,
description="Maximum number of papers to return (default: 20).",
)
sort_by_citation: bool = Field(
default=True,
description="If true, sort by citation count. If false, sort by most recent date.",
)
year: int | None = Field(
default=None,
ge=1900,
le=2100,
description="Optional publication year filter (e.g. 2020).",
)
class TitleSearchInput(BaseSearchInput):
title: str = Field(
...,
min_length=1,
max_length=300,
description="Title keyword or phrase to search.",
)
class FulltextSearchInput(BaseSearchInput):
fulltext: str = Field(
...,
min_length=1,
max_length=300,
description="Full-text keyword or phrase to search in paper body.",
)
class AuthorSearchInput(BaseSearchInput):
author: list[str] = Field(
...,
min_length=1,
max_length=10,
description="Author names to search (e.g. ['Witten, Edward', 'Maldacena, Juan']).",
)
@field_validator("author")
@classmethod
def validate_author_list(cls, value: list[str]) -> list[str]:
cleaned = [item.strip() for item in value]
if any(not item for item in cleaned):
raise ValueError("author entries must be non-empty strings.")
return cleaned
@mcp.tool(
name="inspirehep_search_by_title",
annotations={
"title": "Search INSPIRE-HEP papers by title",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True,
},
)
async def inspirehep_search_by_title(params: TitleSearchInput) -> dict[str, Any]:
"""Search INSPIRE-HEP literature by title."""
query = build_title_query(params.title, params.large_collaboration, params.year)
return await run_search(
query=query,
limit=params.limit,
sort_by_citation=params.sort_by_citation,
)
@mcp.tool(
name="inspirehep_search_by_fulltext",
annotations={
"title": "Search INSPIRE-HEP papers by full text",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True,
},
)
async def inspirehep_search_by_fulltext(params: FulltextSearchInput) -> dict[str, Any]:
"""Search INSPIRE-HEP literature by full text."""
query = build_fulltext_query(params.fulltext, params.large_collaboration, params.year)
return await run_search(
query=query,
limit=params.limit,
sort_by_citation=params.sort_by_citation,
)
@mcp.tool(
name="inspirehep_search_by_author",
annotations={
"title": "Search INSPIRE-HEP papers by author",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True,
},
)
async def inspirehep_search_by_author(params: AuthorSearchInput) -> dict[str, Any]:
"""Search INSPIRE-HEP literature by author."""
query = build_author_query(params.author, params.large_collaboration, params.year)
return await run_search(
query=query,
limit=params.limit,
sort_by_citation=params.sort_by_citation,
)
async def run_search(
*,
query: str,
limit: int,
sort_by_citation: bool,
client: InspireHEPClient | None = None,
) -> dict[str, Any]:
sort = _to_inspire_sort(sort_by_citation)
try:
if client is not None:
search_result = await client.search_literature(query=query, limit=limit, sort=sort)
else:
async with InspireHEPClient() as default_client:
search_result = await default_client.search_literature(
query=query,
limit=limit,
sort=sort,
)
except httpx.HTTPStatusError as exc:
status = exc.response.status_code
message = exc.response.text or "No response body."
raise RuntimeError(
f"INSPIRE API error ({status}): {message[:300]}"
) from exc
except httpx.HTTPError as exc:
raise RuntimeError(f"INSPIRE API request failed: {exc}") from exc
return {
"count": len(search_result.records),
"records": search_result.records,
}
def _to_inspire_sort(sort_by_citation: bool) -> str:
if sort_by_citation:
return "mostcited"
return "mostrecent"
def main() -> None:
mcp.run(transport="stdio")
