diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml
index 3894ee1..47c79ee 100644
--- a/.github/workflows/pythonpackage.yml
+++ b/.github/workflows/pythonpackage.yml
@@ -42,6 +42,10 @@ jobs:
continue-on-error: true
run: |
coverage run -m unittest -v test_module.TestScholarly
+ - name: Run MCP server unit tests
+ # Offline tests; the ones that need the MCP SDK (Python 3.10+) skip themselves on 3.8.
+ run: |
+ coverage run -a -m unittest -v test_mcp
- name: Generate coverage report (macOS only)
if:
"matrix.os == 'macos-latest'"
diff --git a/README.md b/README.md
index d1ac442..7886a8e 100644
--- a/README.md
+++ b/README.md
@@ -51,6 +51,14 @@ This means your code that uses an earlier version of `scholarly` is guaranteed t
```
**Note:** Tor option is unavailable with conda installation.
+- **MCP server** (Python 3.10+):
+
+ To expose `scholarly` to LLM agents and other [Model Context Protocol](https://modelcontextprotocol.io/) clients, install with the `mcp` tag as
+ ```bash
+ pip3 install scholarly[mcp]
+ ```
+ and see [Running as an MCP server](#running-as-an-mcp-server).
+
## Tests
To check if your installation is successful, run the tests by executing the `test_module.py` file as:
@@ -125,6 +133,29 @@ It is therefore recommended to always set up a proxy in the beginning of your ap
The developers use `ScraperAPI` to run the tests in Github Actions.
The developers of `scholarly` are not affiliated with any of the proxy services and do not profit from them. If your favorite service is not supported, please submit an issue or even better, follow it up with a pull request.
+## Running as an MCP server
+
+`scholarly` can run as a [Model Context Protocol](https://modelcontextprotocol.io/) (MCP) server, so that LLM agents and MCP clients (Claude Desktop, IDE agents, agent gateways, ...) can query Google Scholar as a tool.
+
+```bash
+pip3 install scholarly[mcp] # needs Python 3.10+
+scholarly-mcp # stdio transport, for clients that launch the server themselves
+scholarly-mcp --transport streamable-http --host 127.0.0.1 --port 8000
+scholarly-mcp --proxy free # none (default) | free | tor | scraperapi
+```
+
+Example client configuration (Claude Desktop style `mcpServers` entry):
+
+```json
+{"mcpServers": {"scholarly": {"command": "scholarly-mcp", "args": ["--proxy", "free"]}}}
+```
+
+Tools: `search_authors`, `get_author`, `search_publications`, `get_publication`, `get_bibtex`, `get_citing_publications` and `get_related_articles`. All are read-only and take plain names, ids or titles, not Python objects.
+
+Google Scholar blocks automated clients quickly (see the **IMPORTANT** note above), and the server cannot avoid that, so keep `limit` small and use `--proxy`. Credentials are read from the environment, never from flags: `SCRAPER_API_KEY` for `--proxy scraperapi`, and `SCHOLARLY_TOR_SOCKS_PORT`, `SCHOLARLY_TOR_CONTROL_PORT` and `SCHOLARLY_TOR_PASSWORD` for `--proxy tor`.
+The network transports have no authentication, so they bind to `127.0.0.1` by default; do not expose them to untrusted networks.
+See the [documentation](https://scholarly.readthedocs.io/en/stable/mcp.html) for details.
+
## Contributing
We welcome contributions from you.
diff --git a/docs/index.rst b/docs/index.rst
index bd2a44b..d9808e3 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -12,6 +12,7 @@ Welcome to scholarly's documentation!
quickstart
scholarly
+ mcp
Indices and tables
diff --git a/docs/mcp.rst b/docs/mcp.rst
new file mode 100644
index 0000000..0e0eed6
--- /dev/null
+++ b/docs/mcp.rst
@@ -0,0 +1,100 @@
+Running scholarly as an MCP server
+==================================
+
+``scholarly`` can run as a `Model Context Protocol `_ (MCP)
+server, so that LLM agents and other MCP clients can query Google Scholar as a tool.
+The MCP layer is optional and needs Python 3.10 or newer::
+
+ pip3 install scholarly[mcp]
+
+It works with both the 1.x (``FastMCP``) and 2.x (``MCPServer``) generations of the
+`MCP Python SDK `_.
+
+Launching the server
+--------------------
+
+The ``scholarly-mcp`` command (equivalently ``python -m scholarly.mcp``) starts the server::
+
+ scholarly-mcp # stdio transport (default)
+ scholarly-mcp --transport sse --port 8000
+ scholarly-mcp --transport streamable-http --host 127.0.0.1 --port 8000
+ scholarly-mcp --proxy free --log-level INFO
+
+``--transport``
+ ``stdio`` (default) for clients that start the server themselves, or the network
+ transports ``sse`` and ``streamable-http``.
+``--host`` / ``--port``
+ Interface and port of the network transports (default ``127.0.0.1:8000``). The network
+ transports have **no authentication**, so only bind a non-loopback address on a trusted
+ network; the server logs a warning when you do.
+``--proxy``
+ ``none`` (default), ``free``, ``tor`` or ``scraperapi``. This configures the same
+ :class:`ProxyGenerator ` that the Python API uses, and the server
+ refuses to start if the proxy cannot be set up rather than silently running unproxied.
+``--log-level``
+ Logging threshold. Logs are written to ``stderr`` because ``stdout`` carries the protocol
+ on the ``stdio`` transport.
+
+Credentials are read from the environment, never from command line flags, so that they do not
+appear in process listings:
+
+``SCRAPER_API_KEY``
+ API key for ``--proxy scraperapi``.
+``SCHOLARLY_TOR_SOCKS_PORT``, ``SCHOLARLY_TOR_CONTROL_PORT``, ``SCHOLARLY_TOR_PASSWORD``
+ Settings of an already running Tor instance for ``--proxy tor`` (defaults ``9050``, ``9051``
+ and an empty password).
+
+A ``.env`` file is honoured, as for the Python API.
+
+Client configuration
+--------------------
+
+For clients that launch the server (for example Claude Desktop)::
+
+ {
+ "mcpServers": {
+ "scholarly": {
+ "command": "scholarly-mcp",
+ "args": ["--proxy", "free"]
+ }
+ }
+ }
+
+Tools
+-----
+
+All tools are read-only and take plain names, identifiers or titles, because an MCP client
+cannot hold on to ``Author`` or ``Publication`` objects between calls. Results are plain JSON.
+Tools that return a list accept a ``limit`` (at most 50 per call).
+
+``search_authors(name, limit=5)``
+ Search authors by name, optionally with an affiliation, e.g. ``"Marty Banks, Berkeley"``.
+``get_author(scholar_id, sections=None, sortby="citedby", publication_limit=10)``
+ Fetch an author profile by Google Scholar id. ``sections`` is any of ``basics``, ``indices``,
+ ``counts``, ``coauthors``, ``publications`` and ``public_access`` (default: all but
+ ``public_access``). ``publication_limit=0`` fetches every publication.
+``search_publications(query, limit=5, year_low=None, year_high=None, sort_by="relevance", patents=True)``
+ Search publications with a free-text query.
+``get_publication(title)``
+ Full details of the best match for a title.
+``get_bibtex(title)``
+ BibTeX entry of the best match for a title.
+``get_citing_publications(title, limit=10)``
+ Publications that cite the best match for a title.
+``get_related_articles(title, limit=10)``
+ Articles related to the best match for a title.
+
+Calls are executed one at a time. ``scholarly`` keeps a single stateful session and Google
+Scholar penalises bursts, so concurrent requests would only make blocking more likely.
+
+Errors and limitations
+----------------------
+
+Invalid arguments, "no such publication" and Google Scholar refusing to serve the request are
+reported to the client as tool errors with an explanatory message. Unexpected internal errors are
+reported as errors too; SDK 2.x deliberately withholds their details from the client.
+
+Google Scholar sometimes answers a blocked client with a sign-in or CAPTCHA page rather than an
+error, which ``scholarly`` parses as "no results". Empty listings therefore carry a ``note``
+explaining that an empty result can mean "blocked". Keep ``limit`` small, avoid repeating
+identical queries, and use ``--proxy`` (see "Using proxies" in the :doc:`quickstart`).
diff --git a/scholarly/mcp.py b/scholarly/mcp.py
new file mode 100644
index 0000000..073ef4e
--- /dev/null
+++ b/scholarly/mcp.py
@@ -0,0 +1,417 @@
+"""Run ``scholarly`` as a `Model Context Protocol `_ (MCP) server.
+
+This module is optional. Install the MCP SDK with::
+
+ pip install "scholarly[mcp]"
+
+and launch the server with the ``scholarly-mcp`` console script (or
+``python -m scholarly.mcp``)::
+
+ scholarly-mcp # stdio transport (default)
+ scholarly-mcp --transport streamable-http --host 127.0.0.1 --port 8000
+ scholarly-mcp --proxy free # route requests through free proxies
+
+The server exposes thin, stateless wrappers over the public ``scholarly`` API.
+Every tool takes plain JSON-friendly arguments (names, ids, titles) and returns
+plain JSON-friendly results, because MCP clients cannot hold on to Python
+``Author``/``Publication`` objects between calls.
+
+Secrets are read from the environment only (never from command line flags, so
+they do not show up in process listings):
+
+``SCRAPER_API_KEY``
+ API key used by ``--proxy scraperapi``.
+``SCHOLARLY_TOR_SOCKS_PORT`` / ``SCHOLARLY_TOR_CONTROL_PORT`` / ``SCHOLARLY_TOR_PASSWORD``
+ Settings used by ``--proxy tor`` (an already running Tor instance).
+"""
+import argparse
+import asyncio
+import datetime
+import functools
+import inspect
+import itertools
+import logging
+import os
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+from ._proxy_generator import DOSException, MaxTriesExceededException
+
+SERVER_NAME = "scholarly"
+
+SERVER_INSTRUCTIONS = (
+ "Query Google Scholar for authors, publications, citations and BibTeX through the "
+ "`scholarly` library. Google Scholar rate-limits and CAPTCHAs automated clients "
+ "aggressively, so keep `limit` values small and avoid repeating identical queries. "
+ "Lookups by Google Scholar author id (`scholar_id`) are the most reliable; lookups by "
+ "free text return the best match only."
+)
+
+#: Hard cap on results returned by a single tool call. Every ~10 results costs one
+#: extra HTTP request to Google Scholar, so this also bounds the blast radius of a call.
+MAX_RESULTS = 50
+DEFAULT_LIMIT = 5
+
+AUTHOR_SECTIONS = ("basics", "indices", "counts", "coauthors", "publications", "public_access")
+DEFAULT_AUTHOR_SECTIONS = ("basics", "indices", "counts", "coauthors", "publications")
+AUTHOR_SORT_KEYS = ("citedby", "year")
+PUBLICATION_SORT_KEYS = ("relevance", "date")
+
+PROXY_CHOICES = ("none", "free", "tor", "scraperapi")
+TRANSPORT_CHOICES = ("stdio", "sse", "streamable-http")
+LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
+_LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1")
+
+
+class ProxyConfigError(RuntimeError):
+ """Raised when the requested proxy mode cannot be configured."""
+
+
+#: Attached to empty listings. Google Scholar sometimes answers automated clients with a
+#: sign-in/CAPTCHA page instead of an error, which scholarly parses as "no results".
+EMPTY_RESULT_NOTE = (
+ "No results. This may be a genuine absence of matches, but Google Scholar also returns "
+ "empty pages to rate-limited or blocked clients (notably for author searches). Try again "
+ "later, vary the query, or run the server with --proxy."
+)
+
+#: Failures the calling agent can act on (fix the arguments, retry later, use a proxy).
+#: These are reported to the client verbatim. Anything else is treated as a bug and, on
+#: MCP SDK 2.x, deliberately not leaked to the client.
+ANTICIPATED_ERRORS = (ValueError, MaxTriesExceededException, DOSException)
+
+
+def _jsonable(obj: Any) -> Any:
+ """Recursively convert scholarly objects into JSON-serialisable builtins."""
+ if isinstance(obj, Enum):
+ return _jsonable(obj.value)
+ if obj is None or isinstance(obj, (bool, int, float, str)):
+ return obj
+ if isinstance(obj, dict):
+ return {str(k): _jsonable(v) for k, v in obj.items()}
+ if isinstance(obj, (list, tuple, set, frozenset)):
+ return [_jsonable(v) for v in obj]
+ if isinstance(obj, (datetime.datetime, datetime.date)):
+ return obj.isoformat()
+ return str(obj)
+
+
+def _clamp_limit(limit: Optional[int], default: int = DEFAULT_LIMIT) -> int:
+ """Return ``limit`` constrained to ``1..MAX_RESULTS`` (``default`` when ``None``)."""
+ if limit is None:
+ limit = default
+ return max(1, min(int(limit), MAX_RESULTS))
+
+
+def _take(iterator: Any, limit: int) -> List[Any]:
+ """Return at most ``limit`` items from a (possibly ``None``) iterator."""
+ if iterator is None:
+ return []
+ return list(itertools.islice(iterator, limit))
+
+
+def _validate_choice(value: str, choices: tuple, name: str) -> str:
+ if value not in choices:
+ raise ValueError("Invalid %s %r; expected one of: %s" % (name, value, ", ".join(choices)))
+ return value
+
+
+class ScholarlyTools:
+ """The logic behind each MCP tool, kept free of any MCP SDK dependency.
+
+ :param api: the ``scholarly`` API object. Defaults to the module level singleton;
+ tests inject a fake.
+ """
+
+ def __init__(self, api: Any = None):
+ if api is None:
+ import scholarly as _pkg
+ api = _pkg.scholarly
+ self.api = api
+
+ # -- helpers -----------------------------------------------------------------
+ def _find_publication(self, title: str, filled: bool = False) -> Any:
+ """Best-match publication for a title/query, with a clear error if there is none."""
+ if not title or not title.strip():
+ raise ValueError("title must be a non-empty string")
+ try:
+ pub = self.api.search_single_pub(title, filled=filled)
+ except (StopIteration, IndexError):
+ pub = None
+ if not pub:
+ raise ValueError("No publication found on Google Scholar for %r" % title)
+ return pub
+
+ @staticmethod
+ def _listing(query: Any, items: List[Any]) -> Dict[str, Any]:
+ listing = {"query": query, "count": len(items), "results": _jsonable(items)}
+ if not items:
+ listing["note"] = EMPTY_RESULT_NOTE
+ return listing
+
+ # -- tools -------------------------------------------------------------------
+ def search_authors(self, name: str, limit: Optional[int] = None) -> Dict[str, Any]:
+ """Search authors by name (optionally with affiliation, e.g. ``"Marty Banks, Berkeley"``)."""
+ if not name or not name.strip():
+ raise ValueError("name must be a non-empty string")
+ items = _take(self.api.search_author(name), _clamp_limit(limit))
+ return self._listing(name, items)
+
+ def get_author(self, scholar_id: str, sections: Optional[List[str]] = None,
+ sortby: str = "citedby", publication_limit: int = 10) -> Dict[str, Any]:
+ """Fetch an author profile by Google Scholar id and fill the requested sections."""
+ if not scholar_id or not scholar_id.strip():
+ raise ValueError("scholar_id must be a non-empty string")
+ sections = list(sections) if sections else list(DEFAULT_AUTHOR_SECTIONS)
+ for section in sections:
+ _validate_choice(section, AUTHOR_SECTIONS, "section")
+ _validate_choice(sortby, AUTHOR_SORT_KEYS, "sortby")
+ publication_limit = max(0, int(publication_limit))
+ author = self.api.search_author_id(scholar_id)
+ author = self.api.fill(author, sections=sections, sortby=sortby,
+ publication_limit=publication_limit)
+ return _jsonable(author)
+
+ def search_publications(self, query: str, limit: Optional[int] = None,
+ year_low: Optional[int] = None, year_high: Optional[int] = None,
+ sort_by: str = "relevance", patents: bool = True) -> Dict[str, Any]:
+ """Search publications with a free-text Google Scholar query."""
+ if not query or not query.strip():
+ raise ValueError("query must be a non-empty string")
+ _validate_choice(sort_by, PUBLICATION_SORT_KEYS, "sort_by")
+ results = self.api.search_pubs(query, patents=patents, year_low=year_low,
+ year_high=year_high, sort_by=sort_by)
+ return self._listing(query, _take(results, _clamp_limit(limit)))
+
+ def get_publication(self, title: str) -> Dict[str, Any]:
+ """Return the best-matching publication for a title, with full details filled in."""
+ return _jsonable(self._find_publication(title, filled=True))
+
+ def get_bibtex(self, title: str) -> Dict[str, Any]:
+ """Return the BibTeX entry of the best-matching publication for a title."""
+ pub = self._find_publication(title)
+ bibtex = self.api.bibtex(pub)
+ if not bibtex:
+ raise ValueError("Google Scholar returned no BibTeX for %r" % title)
+ return {"title": _jsonable(pub.get("bib", {}).get("title", title)), "bibtex": bibtex}
+
+ def get_citing_publications(self, title: str, limit: Optional[int] = None) -> Dict[str, Any]:
+ """List publications that cite the best-matching publication for a title."""
+ pub = self._find_publication(title)
+ return self._listing(title, _take(self.api.citedby(pub), _clamp_limit(limit, 10)))
+
+ def get_related_articles(self, title: str, limit: Optional[int] = None) -> Dict[str, Any]:
+ """List articles Google Scholar considers related to the best-matching publication."""
+ pub = self._find_publication(title)
+ return self._listing(title, _take(self.api.get_related_articles(pub), _clamp_limit(limit, 10)))
+
+
+# (tool name, description). Each name is also the ScholarlyTools method that implements it.
+TOOL_SPECS = (
+ ("search_authors", "Search Google Scholar authors by name."),
+ ("get_author", "Fetch a Google Scholar author profile (indices, counts, coauthors, publications)."),
+ ("search_publications", "Search Google Scholar publications with a free-text query."),
+ ("get_publication", "Get full details of the best-matching publication for a title."),
+ ("get_bibtex", "Get the BibTeX entry of the best-matching publication for a title."),
+ ("get_citing_publications", "List publications that cite the best-matching publication for a title."),
+ ("get_related_articles", "List articles related to the best-matching publication for a title."),
+)
+
+_DESCRIPTIONS = dict(TOOL_SPECS)
+
+
+def configure_proxy(mode: str, api: Any = None, env: Optional[Dict[str, str]] = None) -> None:
+ """Route ``scholarly`` traffic through the requested proxy.
+
+ :param mode: one of :data:`PROXY_CHOICES`.
+ :raises ProxyConfigError: if the proxy cannot be configured.
+ """
+ _validate_choice(mode, PROXY_CHOICES, "proxy mode")
+ if mode == "none":
+ return
+ env = os.environ if env is None else env
+ if api is None:
+ import scholarly as _pkg
+ api = _pkg.scholarly
+ from ._proxy_generator import ProxyGenerator
+
+ pg = ProxyGenerator()
+ if mode == "free":
+ ok = pg.FreeProxies()
+ elif mode == "scraperapi":
+ key = env.get("SCRAPER_API_KEY")
+ if not key:
+ raise ProxyConfigError("--proxy scraperapi requires the SCRAPER_API_KEY environment variable")
+ ok = pg.ScraperAPI(key)
+ else: # tor
+ try:
+ socks_port = int(env.get("SCHOLARLY_TOR_SOCKS_PORT", 9050))
+ control_port = int(env.get("SCHOLARLY_TOR_CONTROL_PORT", 9051))
+ except ValueError as exc:
+ raise ProxyConfigError("Invalid Tor port in environment: %s" % exc)
+ ok = pg.Tor_External(socks_port, control_port, env.get("SCHOLARLY_TOR_PASSWORD", ""))
+ if not ok:
+ raise ProxyConfigError("Could not set up the %r proxy; scholarly would run unproxied" % mode)
+ api.use_proxy(pg)
+
+
+def _import_server_class():
+ """Return ``(server_class, sdk_major)`` for the installed MCP SDK (1.x or 2.x)."""
+ try:
+ from mcp.server.mcpserver import MCPServer # mcp >= 2
+ return MCPServer, 2
+ except ImportError:
+ pass
+ try:
+ from mcp.server.fastmcp import FastMCP # mcp 1.x
+ return FastMCP, 1
+ except ImportError:
+ raise ImportError(
+ "The MCP SDK is required to run scholarly as an MCP server. "
+ "Install it with: pip install \"scholarly[mcp]\" (requires Python 3.10+)")
+
+
+def _tool_error_class(sdk_major: int) -> Any:
+ """The MCP SDK's ``ToolError``: the only exception type whose message always reaches the client."""
+ if sdk_major == 2:
+ from mcp.server.mcpserver.exceptions import ToolError
+ else:
+ from mcp.server.fastmcp.exceptions import ToolError
+ return ToolError
+
+
+def create_server(api: Any = None, host: str = "127.0.0.1", port: int = 8000) -> Any:
+ """Build the MCP server with all scholarly tools registered.
+
+ ``host``/``port`` only matter for the network transports. With mcp 1.x they are
+ bound to the server here; with mcp 2.x they are passed to :func:`run_server`.
+ """
+ server_cls, sdk_major = _import_server_class()
+ if sdk_major == 1:
+ server = server_cls(SERVER_NAME, instructions=SERVER_INSTRUCTIONS, host=host, port=port)
+ else:
+ server = server_cls(SERVER_NAME, instructions=SERVER_INSTRUCTIONS)
+
+ tools = ScholarlyTools(api)
+ tool_error = _tool_error_class(sdk_major)
+ # Scholarly is a stateful singleton (shared session and proxy) and Google Scholar
+ # punishes bursts, so run every tool on one worker thread, one call at a time.
+ executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="scholarly-mcp")
+
+ # Advertise tools as read-only / open-world when the SDK supports annotations.
+ tool_kwargs = {}
+ if "annotations" in inspect.signature(server.tool).parameters:
+ from mcp.types import ToolAnnotations
+ tool_kwargs["annotations"] = ToolAnnotations(
+ readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True)
+
+ async def _call(method_name: str, *args: Any, **kwargs: Any) -> Dict[str, Any]:
+ method = getattr(tools, method_name)
+ loop = asyncio.get_running_loop()
+ try:
+ return await loop.run_in_executor(executor, functools.partial(method, *args, **kwargs))
+ except ANTICIPATED_ERRORS as exc:
+ raise tool_error(str(exc) or type(exc).__name__) from exc
+
+ @server.tool(name="search_authors", description=_DESCRIPTIONS["search_authors"], **tool_kwargs)
+ async def search_authors(name: str, limit: int = DEFAULT_LIMIT) -> Dict[str, Any]:
+ return await _call("search_authors", name, limit)
+
+ @server.tool(name="get_author", description=_DESCRIPTIONS["get_author"], **tool_kwargs)
+ async def get_author(scholar_id: str, sections: Optional[List[str]] = None,
+ sortby: str = "citedby", publication_limit: int = 10) -> Dict[str, Any]:
+ return await _call("get_author", scholar_id, sections, sortby, publication_limit)
+
+ @server.tool(name="search_publications", description=_DESCRIPTIONS["search_publications"], **tool_kwargs)
+ async def search_publications(query: str, limit: int = DEFAULT_LIMIT,
+ year_low: Optional[int] = None, year_high: Optional[int] = None,
+ sort_by: str = "relevance", patents: bool = True) -> Dict[str, Any]:
+ return await _call("search_publications", query, limit, year_low, year_high, sort_by, patents)
+
+ @server.tool(name="get_publication", description=_DESCRIPTIONS["get_publication"], **tool_kwargs)
+ async def get_publication(title: str) -> Dict[str, Any]:
+ return await _call("get_publication", title)
+
+ @server.tool(name="get_bibtex", description=_DESCRIPTIONS["get_bibtex"], **tool_kwargs)
+ async def get_bibtex(title: str) -> Dict[str, Any]:
+ return await _call("get_bibtex", title)
+
+ @server.tool(name="get_citing_publications", description=_DESCRIPTIONS["get_citing_publications"],
+ **tool_kwargs)
+ async def get_citing_publications(title: str, limit: int = 10) -> Dict[str, Any]:
+ return await _call("get_citing_publications", title, limit)
+
+ @server.tool(name="get_related_articles", description=_DESCRIPTIONS["get_related_articles"], **tool_kwargs)
+ async def get_related_articles(title: str, limit: int = 10) -> Dict[str, Any]:
+ return await _call("get_related_articles", title, limit)
+
+ return server
+
+
+def run_server(server: Any, transport: str = "stdio", host: str = "127.0.0.1", port: int = 8000) -> None:
+ """Run ``server`` on the given transport (blocks until the server stops)."""
+ _validate_choice(transport, TRANSPORT_CHOICES, "transport")
+ if transport == "stdio":
+ server.run(transport="stdio")
+ elif type(server).__name__ == "MCPServer": # mcp >= 2 takes host/port at run time
+ server.run(transport=transport, host=host, port=port)
+ else: # mcp 1.x bound host/port in create_server()
+ server.run(transport=transport)
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="scholarly-mcp",
+ description="Run scholarly (Google Scholar) as a Model Context Protocol server.")
+ parser.add_argument("--transport", choices=TRANSPORT_CHOICES, default="stdio",
+ help="MCP transport (default: stdio)")
+ parser.add_argument("--host", default="127.0.0.1",
+ help="interface to bind for sse/streamable-http (default: 127.0.0.1)")
+ parser.add_argument("--port", type=int, default=8000,
+ help="port to listen on for sse/streamable-http (default: 8000)")
+ parser.add_argument("--proxy", choices=PROXY_CHOICES, default="none",
+ help="proxy for Google Scholar requests; credentials are read from the "
+ "environment (default: none)")
+ parser.add_argument("--log-level", choices=LOG_LEVELS, default="WARNING",
+ help="log level, written to stderr (default: WARNING)")
+ return parser
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+ """Entry point of the ``scholarly-mcp`` console script."""
+ args = build_parser().parse_args(argv)
+
+ # stdout carries the MCP protocol on the stdio transport, so log to stderr only.
+ logging.basicConfig(level=args.log_level, stream=sys.stderr,
+ format="%(asctime)s %(name)s %(levelname)s %(message)s")
+ logging.getLogger("scholarly").setLevel(args.log_level)
+
+ try:
+ server = create_server(host=args.host, port=args.port)
+ except ImportError as exc:
+ print("scholarly-mcp: %s" % exc, file=sys.stderr)
+ return 2
+ try:
+ configure_proxy(args.proxy)
+ except ProxyConfigError as exc:
+ print("scholarly-mcp: %s" % exc, file=sys.stderr)
+ return 2
+
+ if args.transport != "stdio" and args.host not in _LOOPBACK_HOSTS:
+ logging.getLogger(__name__).warning(
+ "Binding %s:%d without authentication; anyone who can reach this port can "
+ "drive Google Scholar queries through your proxy and IP.", args.host, args.port)
+
+ try:
+ run_server(server, args.transport, args.host, args.port)
+ except (KeyboardInterrupt, asyncio.CancelledError):
+ # Ctrl+C is how a server is normally stopped by hand, not a failure: don't print a
+ # traceback and don't report an error status.
+ logging.getLogger(__name__).info("Shutting down")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/setup.py b/setup.py
index b4ea378..1283d0c 100644
--- a/setup.py
+++ b/setup.py
@@ -39,6 +39,13 @@
],
extras_require={
'tor': ['stem'],
+ # The MCP SDK needs Python 3.10+; the core library keeps working on older versions.
+ 'mcp': ['mcp>=1.2; python_version >= "3.10"'],
+ },
+ entry_points={
+ 'console_scripts': [
+ 'scholarly-mcp=scholarly.mcp:main',
+ ],
},
test_suite="test_module.py"
)
diff --git a/test_mcp.py b/test_mcp.py
new file mode 100644
index 0000000..32fea97
--- /dev/null
+++ b/test_mcp.py
@@ -0,0 +1,371 @@
+"""Unit tests for ``scholarly.mcp`` (the optional MCP server).
+
+These tests are fully offline: the scholarly API is replaced by a fake, so no request
+is ever sent to Google Scholar. Tests that need the MCP SDK (Python 3.10+) are skipped
+when it is not installed; everything else also runs on older Python versions.
+"""
+import asyncio
+import datetime
+import importlib.util
+import sys
+import threading
+import time
+import unittest
+from enum import Enum
+from unittest import mock
+
+from scholarly import mcp as smcp
+
+HAS_SDK = sys.version_info >= (3, 10) and importlib.util.find_spec("mcp") is not None
+needs_sdk = unittest.skipUnless(HAS_SDK, "the MCP SDK (pip install 'scholarly[mcp]', Python 3.10+) is not installed")
+
+
+class FakeApi:
+ """Stands in for the ``scholarly`` singleton and records how it was called."""
+
+ def __init__(self):
+ self.calls = []
+ self.proxy = None
+ self.bibtex_text = "@article{x, title={T}}"
+ self.single_pub = {"bib": {"title": "A paper"}, "container_type": "Publication"}
+ self.active = 0
+ self.max_active = 0
+ self._lock = threading.Lock()
+
+ def search_author(self, name):
+ self.calls.append(("search_author", name))
+ with self._lock:
+ self.active += 1
+ self.max_active = max(self.max_active, self.active)
+ time.sleep(0.05)
+ with self._lock:
+ self.active -= 1
+ return iter({"name": "Ada %d" % i, "scholar_id": "ID%d" % i} for i in range(100))
+
+ def search_author_id(self, scholar_id):
+ self.calls.append(("search_author_id", scholar_id))
+ return {"scholar_id": scholar_id, "filled": False}
+
+ def fill(self, obj, sections=None, sortby="citedby", publication_limit=0):
+ self.calls.append(("fill", sections, sortby, publication_limit))
+ return dict(obj, filled=True, sections=list(sections))
+
+ def search_pubs(self, query, **kwargs):
+ self.calls.append(("search_pubs", query, kwargs))
+ return iter({"bib": {"title": "Pub %d" % i}} for i in range(100))
+
+ def search_single_pub(self, title, filled=False):
+ self.calls.append(("search_single_pub", title, filled))
+ if isinstance(self.single_pub, Exception):
+ raise self.single_pub
+ return self.single_pub
+
+ def bibtex(self, pub):
+ return self.bibtex_text
+
+ def citedby(self, pub):
+ return iter({"bib": {"title": "Citing %d" % i}} for i in range(100))
+
+ def get_related_articles(self, pub):
+ return iter({"bib": {"title": "Related %d" % i}} for i in range(100))
+
+ def use_proxy(self, pg):
+ self.proxy = pg
+
+
+class TestHelpers(unittest.TestCase):
+ def test_jsonable_converts_enums_dates_and_unknown_objects(self):
+ class Color(str, Enum):
+ RED = "red"
+
+ class Odd:
+ def __str__(self):
+ return "odd"
+
+ data = {"c": Color.RED, "when": datetime.date(2026, 10, 30), "t": (1, 2), "s": {3},
+ "n": None, "o": Odd(), 5: [Color.RED]}
+ self.assertEqual(smcp._jsonable(data),
+ {"c": "red", "when": "2026-10-30", "t": [1, 2], "s": [3],
+ "n": None, "o": "odd", "5": ["red"]})
+
+ def test_clamp_limit(self):
+ self.assertEqual(smcp._clamp_limit(None), smcp.DEFAULT_LIMIT)
+ self.assertEqual(smcp._clamp_limit(None, 10), 10)
+ self.assertEqual(smcp._clamp_limit(0), 1)
+ self.assertEqual(smcp._clamp_limit(-4), 1)
+ self.assertEqual(smcp._clamp_limit(10 ** 6), smcp.MAX_RESULTS)
+
+ def test_take_handles_none_and_short_iterators(self):
+ self.assertEqual(smcp._take(None, 3), [])
+ self.assertEqual(smcp._take(iter([1, 2]), 5), [1, 2])
+ self.assertEqual(smcp._take(iter(range(10)), 3), [0, 1, 2])
+
+
+class TestScholarlyTools(unittest.TestCase):
+ def setUp(self):
+ self.api = FakeApi()
+ self.tools = smcp.ScholarlyTools(self.api)
+
+ def test_search_authors_respects_limit(self):
+ out = self.tools.search_authors("Ada Lovelace", 3)
+ self.assertEqual((out["query"], out["count"], len(out["results"])), ("Ada Lovelace", 3, 3))
+
+ def test_search_authors_caps_limit_at_max_results(self):
+ self.assertEqual(self.tools.search_authors("Ada", 10 ** 6)["count"], smcp.MAX_RESULTS)
+
+ def test_search_authors_rejects_blank_name(self):
+ with self.assertRaises(ValueError):
+ self.tools.search_authors(" ")
+
+ def test_get_author_uses_default_sections_and_publication_limit(self):
+ out = self.tools.get_author("ABC123")
+ self.assertTrue(out["filled"])
+ self.assertEqual(out["sections"], list(smcp.DEFAULT_AUTHOR_SECTIONS))
+ self.assertIn(("fill", list(smcp.DEFAULT_AUTHOR_SECTIONS), "citedby", 10), self.api.calls)
+
+ def test_get_author_validates_inputs(self):
+ with self.assertRaises(ValueError):
+ self.tools.get_author("ABC", sections=["nonsense"])
+ with self.assertRaises(ValueError):
+ self.tools.get_author("ABC", sortby="alphabet")
+ with self.assertRaises(ValueError):
+ self.tools.get_author("")
+
+ def test_get_author_zero_publication_limit_means_unlimited_and_negative_is_clamped(self):
+ self.tools.get_author("ABC", publication_limit=-5)
+ self.assertEqual(self.api.calls[-1][3], 0)
+
+ def test_search_publications_forwards_filters(self):
+ out = self.tools.search_publications("deep learning", 2, year_low=2020, year_high=2024,
+ sort_by="date", patents=False)
+ self.assertEqual(out["count"], 2)
+ _, query, kwargs = [c for c in self.api.calls if c[0] == "search_pubs"][0]
+ self.assertEqual(query, "deep learning")
+ self.assertEqual(kwargs, {"patents": False, "year_low": 2020, "year_high": 2024, "sort_by": "date"})
+
+ def test_search_publications_validates_sort_by(self):
+ with self.assertRaises(ValueError):
+ self.tools.search_publications("q", sort_by="newest")
+
+ def test_missing_publication_becomes_a_clear_error(self):
+ for missing in (None, {}, IndexError("empty"), StopIteration()):
+ self.api.single_pub = missing
+ with self.assertRaisesRegex(ValueError, "No publication found"):
+ self.tools.get_publication("nothing")
+
+ def test_get_publication_asks_for_a_filled_record(self):
+ self.tools.get_publication("A paper")
+ self.assertIn(("search_single_pub", "A paper", True), self.api.calls)
+
+ def test_get_bibtex(self):
+ out = self.tools.get_bibtex("A paper")
+ self.assertEqual(out, {"title": "A paper", "bibtex": "@article{x, title={T}}"})
+
+ def test_get_bibtex_errors_when_scholar_returns_nothing(self):
+ self.api.bibtex_text = None
+ with self.assertRaisesRegex(ValueError, "no BibTeX"):
+ self.tools.get_bibtex("A paper")
+
+ def test_citing_and_related_listings(self):
+ self.assertEqual(self.tools.get_citing_publications("A paper", 4)["count"], 4)
+ self.assertEqual(self.tools.get_related_articles("A paper", 2)["count"], 2)
+
+ def test_empty_listings_explain_that_blocking_is_possible(self):
+ self.api.search_author = lambda name: iter([])
+ out = self.tools.search_authors("Nobody")
+ self.assertEqual((out["count"], out["results"]), (0, []))
+ self.assertIn("blocked", out["note"])
+ self.assertNotIn("note", smcp.ScholarlyTools(FakeApi()).search_authors("Ada", 1))
+
+ def test_citing_and_related_tolerate_none(self):
+ self.api.citedby = lambda pub: None
+ self.api.get_related_articles = lambda pub: None
+ self.assertEqual(self.tools.get_citing_publications("A paper")["results"], [])
+ self.assertEqual(self.tools.get_related_articles("A paper")["results"], [])
+
+
+class TestConfigureProxy(unittest.TestCase):
+ def setUp(self):
+ self.api = FakeApi()
+ patcher = mock.patch("scholarly._proxy_generator.ProxyGenerator")
+ self.PG = patcher.start()
+ self.addCleanup(patcher.stop)
+ self.pg = self.PG.return_value
+
+ def test_none_is_a_no_op(self):
+ smcp.configure_proxy("none", self.api, env={})
+ self.PG.assert_not_called()
+ self.assertIsNone(self.api.proxy)
+
+ def test_unknown_mode_is_rejected(self):
+ with self.assertRaises(ValueError):
+ smcp.configure_proxy("carrier-pigeon", self.api, env={})
+
+ def test_free(self):
+ self.pg.FreeProxies.return_value = True
+ smcp.configure_proxy("free", self.api, env={})
+ self.assertIs(self.api.proxy, self.pg)
+
+ def test_failure_to_configure_is_an_error_not_a_silent_fallback(self):
+ self.pg.FreeProxies.return_value = False
+ with self.assertRaises(smcp.ProxyConfigError):
+ smcp.configure_proxy("free", self.api, env={})
+ self.assertIsNone(self.api.proxy)
+
+ def test_scraperapi_reads_key_from_env_only(self):
+ self.pg.ScraperAPI.return_value = True
+ smcp.configure_proxy("scraperapi", self.api, env={"SCRAPER_API_KEY": "k"})
+ self.pg.ScraperAPI.assert_called_once_with("k")
+
+ def test_scraperapi_without_key(self):
+ with self.assertRaisesRegex(smcp.ProxyConfigError, "SCRAPER_API_KEY"):
+ smcp.configure_proxy("scraperapi", self.api, env={})
+
+ def test_tor_uses_defaults_and_env_overrides(self):
+ self.pg.Tor_External.return_value = True
+ smcp.configure_proxy("tor", self.api, env={})
+ self.pg.Tor_External.assert_called_with(9050, 9051, "")
+ smcp.configure_proxy("tor", self.api, env={"SCHOLARLY_TOR_SOCKS_PORT": "1",
+ "SCHOLARLY_TOR_CONTROL_PORT": "2",
+ "SCHOLARLY_TOR_PASSWORD": "pw"})
+ self.pg.Tor_External.assert_called_with(1, 2, "pw")
+
+ def test_tor_rejects_bad_port(self):
+ with self.assertRaises(smcp.ProxyConfigError):
+ smcp.configure_proxy("tor", self.api, env={"SCHOLARLY_TOR_SOCKS_PORT": "abc"})
+
+
+class TestCli(unittest.TestCase):
+ def test_defaults(self):
+ args = smcp.build_parser().parse_args([])
+ self.assertEqual((args.transport, args.host, args.port, args.proxy, args.log_level),
+ ("stdio", "127.0.0.1", 8000, "none", "WARNING"))
+
+ def test_rejects_unknown_transport(self):
+ with mock.patch("sys.stderr"), self.assertRaises(SystemExit):
+ smcp.build_parser().parse_args(["--transport", "carrier-pigeon"])
+
+ def test_main_reports_missing_sdk_with_exit_code_2(self):
+ with mock.patch.object(smcp, "_import_server_class", side_effect=ImportError("install mcp")), \
+ mock.patch("sys.stderr") as err:
+ self.assertEqual(smcp.main([]), 2)
+ self.assertIn("install mcp", "".join(c.args[0] for c in err.write.call_args_list))
+
+ def test_main_reports_proxy_failure_with_exit_code_2(self):
+ with mock.patch.object(smcp, "create_server"), \
+ mock.patch.object(smcp, "configure_proxy", side_effect=smcp.ProxyConfigError("nope")), \
+ mock.patch.object(smcp, "run_server") as run, mock.patch("sys.stderr"):
+ self.assertEqual(smcp.main(["--proxy", "free"]), 2)
+ run.assert_not_called()
+
+ def test_main_wires_arguments_through(self):
+ with mock.patch.object(smcp, "create_server") as create, \
+ mock.patch.object(smcp, "configure_proxy") as proxy, \
+ mock.patch.object(smcp, "run_server") as run:
+ code = smcp.main(["--transport", "sse", "--host", "127.0.0.1", "--port", "9001", "--proxy", "free"])
+ self.assertEqual(code, 0)
+ create.assert_called_once_with(host="127.0.0.1", port=9001)
+ proxy.assert_called_once_with("free")
+ run.assert_called_once_with(create.return_value, "sse", "127.0.0.1", 9001)
+
+ def test_main_warns_when_binding_a_non_loopback_address(self):
+ with mock.patch.object(smcp, "create_server"), mock.patch.object(smcp, "configure_proxy"), \
+ mock.patch.object(smcp, "run_server"), self.assertLogs("scholarly.mcp", "WARNING") as logs:
+ smcp.main(["--transport", "streamable-http", "--host", "0.0.0.0"])
+ self.assertIn("without authentication", logs.output[0])
+
+ def test_main_treats_ctrl_c_as_a_clean_shutdown(self):
+ for stop in (KeyboardInterrupt(), asyncio.CancelledError()):
+ with mock.patch.object(smcp, "create_server"), mock.patch.object(smcp, "configure_proxy"), \
+ mock.patch.object(smcp, "run_server", side_effect=stop):
+ self.assertEqual(smcp.main([]), 0, repr(stop))
+
+ def test_run_server_dispatch_for_both_sdk_generations(self):
+ class MCPServer: # mcp >= 2: host/port are given at run time
+ def run(self, **kwargs):
+ self.kwargs = kwargs
+
+ class FastMCP: # mcp 1.x: host/port were bound at construction
+ def run(self, **kwargs):
+ self.kwargs = kwargs
+
+ v2, v1 = MCPServer(), FastMCP()
+ smcp.run_server(v2, "sse", "127.0.0.1", 9)
+ smcp.run_server(v1, "sse", "127.0.0.1", 9)
+ self.assertEqual(v2.kwargs, {"transport": "sse", "host": "127.0.0.1", "port": 9})
+ self.assertEqual(v1.kwargs, {"transport": "sse"})
+ smcp.run_server(v2, "stdio")
+ self.assertEqual(v2.kwargs, {"transport": "stdio"})
+ with self.assertRaises(ValueError):
+ smcp.run_server(v2, "carrier-pigeon")
+
+
+def _payload(result):
+ """Extract the structured tool output from either SDK generation's call_tool result."""
+ structured = result[1] if isinstance(result, tuple) else result.structured_content
+ return structured["result"]
+
+
+@needs_sdk
+class TestServer(unittest.IsolatedAsyncioTestCase):
+ def setUp(self):
+ self.api = FakeApi()
+ self.server = smcp.create_server(self.api)
+
+ async def test_registers_every_tool(self):
+ tools = await self.server.list_tools()
+ self.assertEqual([t.name for t in tools], [name for name, _ in smcp.TOOL_SPECS])
+ for tool in tools:
+ self.assertTrue(tool.description)
+
+ async def test_tools_are_advertised_as_read_only(self):
+ for tool in await self.server.list_tools():
+ if tool.annotations is None: # very old SDKs without annotations
+ continue
+ hints = tool.annotations.model_dump(by_alias=True)
+ self.assertTrue(hints["readOnlyHint"])
+ self.assertTrue(hints["openWorldHint"])
+ self.assertFalse(hints["destructiveHint"])
+
+ async def test_call_tool_end_to_end(self):
+ out = _payload(await self.server.call_tool("search_authors", {"name": "Ada", "limit": 2}))
+ self.assertEqual((out["query"], out["count"]), ("Ada", 2))
+ out = _payload(await self.server.call_tool("get_bibtex", {"title": "A paper"}))
+ self.assertEqual(out["bibtex"], self.api.bibtex_text)
+
+ async def _error_text(self, tool, arguments):
+ """Call a tool that must fail; return whatever error text the client would see."""
+ try:
+ result = await self.server.call_tool(tool, arguments)
+ except Exception as exc: # mcp raises ...
+ return str(exc)
+ # ... or returns an error result
+ self.assertTrue(getattr(result, "is_error", False), "tool unexpectedly succeeded: %r" % (result,))
+ return str(result)
+
+ async def test_tool_errors_reach_the_client(self):
+ self.api.single_pub = None
+ self.assertIn("No publication found", await self._error_text("get_publication", {"title": "nothing"}))
+
+ async def test_bad_arguments_reach_the_client(self):
+ text = await self._error_text("search_publications", {"query": "q", "sort_by": "newest"})
+ self.assertIn("Invalid sort_by", text)
+
+ async def test_google_scholar_blocking_is_explained_to_the_client(self):
+ self.api.single_pub = smcp.MaxTriesExceededException("Cannot fetch from Google Scholar.")
+ self.assertIn("Cannot fetch from Google Scholar", await self._error_text("get_bibtex", {"title": "x"}))
+
+ async def test_unexpected_crashes_are_reported_as_errors_not_successes(self):
+ self.api.single_pub = RuntimeError("internal detail /secret/path")
+ text = await self._error_text("get_bibtex", {"title": "x"})
+ self.assertIn("get_bibtex", text) # an error was reported; SDK 2.x withholds the detail
+
+ async def test_calls_are_serialised_onto_one_worker_thread(self):
+ await asyncio.gather(*(self.server.call_tool("search_authors", {"name": "Ada %d" % i})
+ for i in range(4)))
+ self.assertEqual(self.api.max_active, 1)
+ self.assertEqual(len([c for c in self.api.calls if c[0] == "search_author"]), 4)
+
+
+if __name__ == "__main__":
+ unittest.main()