"""Wikipedia MCP Server implementation."""
import asyncio
from typing import Optional
from mcp.server.fastmcp import FastMCP
from .models import (
WikipediaSearchResult,
WikipediaArticle,
WikipediaSummary,
RelatedArticles,
)
from .wikipedia_client import WikipediaClient
# Global variables
mcp: Optional[FastMCP] = None
wikipedia_client: Optional[WikipediaClient] = None
def get_mcp_server() -> FastMCP:
"""Get or create the MCP server instance."""
global mcp
if mcp is None:
mcp = FastMCP("Wikipedia Research Assistant")
_register_tools()
return mcp
async def get_wikipedia_client() -> WikipediaClient:
"""Get or create the Wikipedia client."""
global wikipedia_client
if wikipedia_client is None:
wikipedia_client = WikipediaClient()
return wikipedia_client
def _register_tools():
"""Register all MCP tools."""
mcp_server = get_mcp_server()
@mcp_server.tool()
async def search_wikipedia(
query: str,
limit: int = 5,
language: str = "en"
) -> WikipediaSearchResult:
"""Search Wikipedia for articles matching the query.
This is the starting point for research - use when you need to find
articles about a topic but don't know exact titles.
Args:
query: Search terms (e.g. 'quantum computing', 'Napoleon Bonaparte')
limit: Number of results to return (1-10, default: 5)
language: Wikipedia language code (default: 'en')
Returns:
Search results with titles, snippets, and URLs
"""
# Validate inputs
if not query.strip():
raise ValueError("Query cannot be empty")
if not (1 <= limit <= 10):
raise ValueError("Limit must be between 1 and 10")
client = await get_wikipedia_client()
return await client.search_wikipedia(query, limit, language)
@mcp_server.tool()
async def get_article(
title: str,
language: str = "en"
) -> WikipediaArticle:
"""Retrieve the full content of a Wikipedia article.
Use this when you know the article title and need the complete content
for detailed analysis or to answer specific questions.
Args:
title: Exact article title (e.g. 'Artificial Intelligence')
language: Wikipedia language code (default: 'en')
Returns:
Full article content with metadata and sections
"""
# Validate inputs
if not title.strip():
raise ValueError("Title cannot be empty")
client = await get_wikipedia_client()
return await client.get_article(title, language)
@mcp_server.tool()
async def get_summary(
title: str,
language: str = "en"
) -> WikipediaSummary:
"""Get a concise summary of a Wikipedia article.
Use this for quick understanding or when you need just the key facts
without the full article content.
Args:
title: Article title to summarize
language: Wikipedia language code (default: 'en')
Returns:
Article summary with key facts
"""
# Validate inputs
if not title.strip():
raise ValueError("Title cannot be empty")
client = await get_wikipedia_client()
return await client.get_summary(title, language)
@mcp_server.tool()
async def find_related(
title: str,
limit: int = 5,
language: str = "en"
) -> RelatedArticles:
"""Find articles related to the given article.
Use this to expand research, find connected topics, or discover
relevant context around your main research subject.
Args:
title: Article title to find related articles for
limit: Number of related articles to return (1-10, default: 5)
language: Wikipedia language code (default: 'en')
Returns:
Related articles with relationship types
"""
# Validate inputs
if not title.strip():
raise ValueError("Title cannot be empty")
if not (1 <= limit <= 10):
raise ValueError("Limit must be between 1 and 10")
client = await get_wikipedia_client()
return await client.find_related(title, limit, language)
async def cleanup():
"""Cleanup resources when the server shuts down."""
global wikipedia_client
if wikipedia_client:
await wikipedia_client.__aexit__(None, None, None)
wikipedia_client = None
def main():
"""Main entry point for the Wikipedia MCP server."""
try:
mcp_server = get_mcp_server()
mcp_server.run()
except KeyboardInterrupt:
print("\nShutting down Wikipedia MCP server...")
finally:
# Cleanup
if wikipedia_client:
asyncio.run(cleanup())
if __name__ == "__main__":
main()