Model Context Protocol - Server and Configuration Files (Claude Desktop Integration)

claude_desktop_config.json

{
  "mcpServers": {
    "expertise-search": {
      "command": "/ADD-FILEPATH-TO-YOUR-PYTHON-VIRTUAL-ENVIRONMENT/.venv/bin/python",
      "args": ["/ADD-FILEPATH-TO-YOUR-LOCAL-MCP-SERVER-FILE/server.py"],
      "env": {
        "PYTHONPATH": "/ADD-FILEPATH-TO-YOUR-LOCAL-MCP-SERVER-DIRECTORY",
        "GOOGLE_API_KEY": "ADD-YOUR-OWN-API-KEY"
      }
    }
  }
}

server.py

#!/usr/bin/env python3
# /// script
# dependencies = [
#     "mcp",
#     "httpx",
#     "python-dotenv",
# ]
# ///

"""
Simplified MCP Server for Wikipedia and MSU Expertise Search
"""

import json
import logging
import os
from dotenv import load_dotenv
import sys
from typing import Any, Dict, Optional

import asyncio
import json
from mcp.client.stdio import StdioServerParameters, stdio_client

import httpx
from mcp.server.fastmcp import FastMCP

# Loads the .env file
load_dotenv()

# Configure logging with more detailed format
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stderr)  # Send logs to stderr to avoid interfering with MCP stdio
    ]
)
logger = logging.getLogger(__name__)

# Initialize MCP server
mcp = FastMCP("expertise-search")
"""
# routine for passing external system prompt to server
try:
    with open("system-prompt.md", "r") as f:
        system_prompt_content = f.read().strip()
    mcp = FastMCP("expertise-search", system_prompt=system_prompt_content)
except FileNotFoundError:
    logger.error("System prompt file not found. Starting without a system prompt.")
    mcp = FastMCP("expertise-search")
"""
# Global HTTP client
client: Optional[httpx.AsyncClient] = None

async def get_client() -> httpx.AsyncClient:
    """Get or create HTTP client"""
    global client
    if client is None:
        logger.info("Creating new HTTP client")
        client = httpx.AsyncClient(timeout=60.0)
    return client

@mcp.tool()
async def wikipedia_search(query: str) -> str:
    """
    Fetches a summary from Wikipedia based on a search query.

    Args:
      query: The search query for Wikipedia.

    Returns:
      A string containing the summary or an error message.
    """
    logger.info(f"Wikipedia search requested for query: '{query}'")
    
    try:
        http_client = await get_client()
        response = await http_client.get("https://en.wikipedia.org/w/api.php", params={
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        })

        if not response.is_success:
            error_msg = f"Error fetching Wikipedia data: {response.status_code} {response.reason_phrase}"
            logger.error(error_msg)
            return error_msg

        try:
            data = response.json()
        except json.JSONDecodeError as e:
            error_msg = f"Invalid JSON response from Wikipedia: {e}"
            logger.error(error_msg)
            return error_msg

        if not data.get("query", {}).get("search", []):
            result = f"No Wikipedia results found for '{query}'"
            logger.info(result)
            return result

        snippet = data["query"]["search"][0]["snippet"]
        logger.info(f"Wikipedia search completed successfully for '{query}'")
        return snippet
        
    except Exception as e:
        error_msg = f"Unexpected error in wikipedia_search: {str(e)}"
        logger.error(error_msg)
        return error_msg

@mcp.tool()
async def search_msu_expertise(query: str) -> str:
    """
    Performs a search for MSU expertise and researcher interests using a custom search engine.

    Search and response format requirements:
    - Use simple searches with unigrams or bigrams.
    - Provide full names, titles, departments, and URLs in Markdown format.
    - Provide confidence score (0-100%) and brief reasoning.
    - Explain reasons for matching people to the query based on their research.

    Args:
      query: The search query.

    Returns:
      A JSON string containing a list of search results, where each result
      contains 'title', 'link', and 'snippet'.
    """
    logger.info(f"MSU expertise search requested for query: '{query}'")
    
    try:
        http_client = await get_client()
        projectID = 'expertise-finder-351604'
        agentID = 'msu-expertise-finder-ai_1731776759855'
        apiKey = os.getenv('GOOGLE_API_KEY')
        if not apiKey:
            error_msg = "GOOGLE_API_KEY environment variable not found"
            logger.error(error_msg)
            return error_msg
        
        url = f"https://discoveryengine.googleapis.com/v1/projects/{projectID}/locations/global/collections/default_collection/engines/{agentID}/servingConfigs/default_search:searchLite?key={apiKey}"
        
        #using searchLite config here: https://cloud.google.com/generative-ai-app-builder/docs/reference/rest/v1/projects.locations.collections.engines.servingConfigs/searchLite
        data = {
            "servingConfig": f"projects/{projectID}/locations/global/collections/default_collection/engines/{agentID}/servingConfigs/default_search",
            "query": query,
            "pageSize": 15,
            "contentSearchSpec": {
                "summarySpec": {
                    "summaryResultCount": 3,
                    "includeCitations": True
                },
                "extractiveContentSpec": {
                    "maxExtractiveAnswerCount": 1
                }
            }
        }

        response = await http_client.post(url, json=data)

        if not response.is_success:
            error_msg = f"Error fetching Google Search results: {response.status_code} {response.reason_phrase}"
            logger.error(error_msg)
            return error_msg

        try:
            decoded_response = response.json()
        except json.JSONDecodeError as e:
            error_msg = f"Invalid JSON response from Google Search: {e}"
            logger.error(error_msg)
            return error_msg

        # Extract and format search results
        results = []
        for result in decoded_response.get('results', []):
            try:
                extracted_result = {
                    'title': result['document']['derivedStructData']['title'],
                    'link': result['document']['derivedStructData']['link'],
                    'snippet': result['document']['derivedStructData']['snippets'][0]['snippet']
                }
                results.append(extracted_result)
            except KeyError as e:
                logger.warning(f"Missing expected field in result: {e}")
                continue

        logger.info(f"MSU expertise search completed successfully for '{query}', found {len(results)} results")
        return json.dumps(results, indent=2)
        
    except Exception as e:
        error_msg = f"Unexpected error in search_msu_expertise: {str(e)}"
        logger.error(error_msg)
        return error_msg

def main():
    """Main entry point for the MCP server"""
    logger.info("Starting MCP server...")
    logger.info("This is an MCP server that communicates via stdio (not HTTP)")
    logger.info("Available tools: wikipedia_search, search_msu_expertise")

    # Add this line to force flush logs
    for handler in logging.root.handlers:
        if isinstance(handler, logging.StreamHandler):
            handler.flush()

    try:
        mcp.run()
    except KeyboardInterrupt:
        logger.info("Server stopped by user")
    except Exception as e:
        logger.error(f"Server error: {e}")
    finally:
        logger.info("MCP server finished running.")

if __name__ == "__main__":
    main()