Add SMD MCP Server implementation and utilities

Implemented the SMD MCP Server with functionality for AI-driven research and article retrieval via the MCP protocol. Added utilities for prompt creation and enhanced dependency configuration. This serves as a foundation for handling SMD-related queries and data processing.
This commit is contained in:
ThomasTaroni 2025-05-31 17:08:46 +02:00
parent 47c036a973
commit 3b75d04f32
4 changed files with 181 additions and 2 deletions

View file

@ -1,6 +1,6 @@
# GPT Researcher dependencies
gpt-researcher>=0.12.16
python-dotenv
python-dotenv~=1.1.0
# MCP dependencies
mcp>=1.6.0
@ -9,4 +9,5 @@ uvicorn>=0.23.2
pydantic>=2.3.0
# Utility dependencies
loguru>=0.7.0
loguru>=0.7.0
requests~=2.32.3

View file

@ -0,0 +1,8 @@
"""
SMD MCP Server
This module provides an MCP server implementation for SMD,
allowing AI assistants to perform searches and generate reports via the MCP protocol.
"""
__version__ = "0.1.0"

View file

@ -0,0 +1,121 @@
"""
SMD Researcher MCP Server
This script implements an MCP server for SMD Researcher, allowing AI assistants
to conduct research and generate reports via the MCP protocol.
"""
import os
import logging
import requests
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
# Load environment variables
load_dotenv()
from utils import (
create_research_prompt
)
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s][%(levelname)s] - %(message)s',
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("SMD Researcher")
@mcp.tool()
async def smd_detail_article(article_id: str) -> dict:
"""
Get the Details of an article found by the SMD Research.
Use this tool when you need the full content of an article.
Args:
article_id: The ID of the article to get the details for
Returns:
String containing the details of the article
"""
url = f"https://api.smd.ch/api/documents/{article_id}"
headers = {
"authorization": f"Bearer {os.getenv('SWISSDOX_BEARER_TOKEN', '')}",
"content-type": "application/json",
}
payload = {"filters": [], "pagination": {"pageSize": 1, "currentPage": 1}}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 200:
return response.json()
else:
return {
"message": response.text
}
@mcp.tool()
async def smd_research(query: dict) -> dict:
"""
Execute a deep search on a given query using SMD Researcher.
Use this tool when you need research on a topic.
Args:
query: The research query or topic
Returns:
String containing research status, ID, and the actual research context
"""
url = "https://api.swissdox.ch/api/documents/search"
headers = {
"authorization": f"Bearer {os.getenv('SWISSDOX_BEARER_TOKEN', '')}",
"content-type": "application/json",
}
response = requests.post(url, headers=headers, json=query)
if response.status_code == 200:
return response.json()
else:
return {
"message": response.text
}
@mcp.prompt()
def research_query(search_query: str = "Bundesrat", date_from: str = "2024-05-30T22:00:00.000Z", date_to: str = "2025-05-31T21:59:59.999Z") -> dict:
"""
Create a research query prompt for SMD Researcher.
Args:
search_query: The SMD search query, there are Logical Operators available (AND, OR, NOT) and for a excact match use "+" before the word. For excluding use "-" before the word. For queries with multiple words use quotes.
date_from: The date to start research from, in the format YYYY-MM-DDTHH:MM:SS.SSSZ
date_to: The date to end research at, in the format YYYY-MM-DDTHH:MM:SS.SSSZ
Returns:
A formatted prompt for research
"""
return create_research_prompt(search_query,date_from,date_to)
def run_server():
"""Run the MCP server using FastMCP's built-in event loop handling."""
# Add startup message
logger.info("Starting GPT Researcher MCP Server...")
print("🚀 GPT Researcher MCP Server starting... Check researcher_mcp_server.log for details")
# Let FastMCP handle the event loop
try:
mcp.run("sse")
# Note: If we reach here, the server has stopped
logger.info("MCP Server has stopped")
except Exception as e:
logger.error(f"Error running MCP server: {str(e)}")
print(f"❌ MCP Server error: {str(e)}")
return
print("✅ MCP Server stopped")
if __name__ == "__main__":
# Use the non-async approach to avoid asyncio nesting issues
run_server()

View file

@ -0,0 +1,49 @@
"""
SMD Researcher MCP Server Utilities
This module provides utility functions and helpers for the SMD Researcher MCP Server.
"""
import sys
from loguru import logger
# Configure logging for console only (no file logging)
logger.configure(handlers=[{"sink": sys.stderr, "level": "INFO"}])
def create_research_prompt(search_query: str = "Bundesrat", date_from: str = "2024-05-30T22:00:00.000Z", date_to: str = "2025-05-31T21:59:59.999Z") -> dict:
"""
Create a research query prompt for SMD Researcher.
Args:
search_query: The SMD search query, there are Logical Operators available (AND, OR, NOT) and for a excact match use "+" before the word. For excluding use "-" before the word. For queries with multiple words use quotes.
date_from: The date to start research from, in the format YYYY-MM-DDTHH:MM:SS.SSSZ
date_to: The date to end research at, in the format YYYY-MM-DDTHH:MM:SS.SSSZ
Returns:
A formatted prompt for research as a Python dictionary.
"""
return {
"sort": {
"field": "score",
"direction": "desc"
},
"filters": [
{
"field": "datetime",
"value": [
date_from,
date_to
]
},
{
"field": "query_text",
"value": [search_query]
}
],
"exact": False,
"pagination": {
"pageSize": 10,
"currentPage": 1
},
"onlyResults": True
}