server.py

from typing import Any, List, Literal, Dict, Optional
import httpx
import traceback
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("patent-problematic-generator-helper")

# API used
ARXIV_BASE = "https://om4r932-arxiv.hf.space"
DUCKDUCKGO_BASE = "https://ychkhan-ptt-endpoints.hf.space"
DOC3GPPFINDER_BASE = "https://organizedprogrammers-3gppdocfinder.hf.space"

# Request function

async def post_data_to_api(url, **kwargs):
    data = dict(kwargs)
    if data is None or data == {}:
        return (None, "")
    headers = {"Accept": "application/json"}
    async with httpx.AsyncClient(verify=False, timeout=180) as client:
        try:
            response = await client.post(url, headers=headers, json=data)
            print(response)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            traceback.print_exception(e)
            return (None, e)
        
async def fake_post_data_to_api(url, **kwargs):
    params = dict(kwargs)
    if params is None or params == {}:
        return (None, "")
    headers = {"Accept": "application/json"}
    async with httpx.AsyncClient(verify=False, timeout=180) as client:
        try:
            response = await client.post(url, headers=headers, params=params)
            print(response)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            traceback.print_exception(e)
            return (None, e)

async def get_data_from_api(url):
    headers = {"Accept": "application/json"}
    async with httpx.AsyncClient(verify=False, timeout=180) as client:
        try:
            response = await client.get(url, headers=headers)
            print(response)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            traceback.print_exception(e)
            return (None, e)

# Tools

# arXiv

@mcp.tool()
async def get_arxiv_publications(keywords: str, limit: int):
    """
    Search arXiv publications based on keywords and a limit of documents printed
    Arguments available: keywords: string [mandatory], limit: integer [mandatory, default = 5]
    """
    endpoint = ARXIV_BASE + "/search"
    data = await post_data_to_api(endpoint, keyword=keywords, limit=limit)
    if isinstance(data, tuple) and data[0] is None:
        return f"An error has occured while getting publications: {data[1]}"
    if data["error"]:
        return data["message"]
    if len(data) < 1:
        return "No publications has been found"
    
    results = data["message"]
    output = []
    for pub, metadata in results.items():
        output.append(f"arXiv pub ID: {pub}\nTitle: {metadata['title']}\nAuthors: {metadata['authors']}\nPublished on: {metadata['date']}\nAbstract: {metadata['abstract']}\nPDF URL: {metadata['pdf']}\n")
    
    return "-\n".join(output)

# 3GPP Doc Finder

@mcp.tool()
async def get_document_url(doc_id: str, release: int = None):
    """
    Find 3GPP document (TSG docs, specifications or workshop files) only by their ID [note that it will only work with keywords] (and release if it's a specification only) and return their position via a URL (and a scope if it's a specification)
    Arguments available: doc_id: string [mandatory], release: integer [optional for every case]
    """
    endpoint = DOC3GPPFINDER_BASE + "/find"
    data = await post_data_to_api(endpoint, doc_id=doc_id, release=release)
    if isinstance(data, tuple) and data[0] is None:
        return f"An error while searching publications: {data[1]}"
    output = f'Document ID: {doc_id}\nURL: {data.get("url", "Not found !")}'
    output += f'\nScope: {data["scope"]}' if data.get("scope", None) is not None else ""
    return output
    
@mcp.tool()
async def search_specs(keywords: str, limit: int):
    """
    Search 3GPP specifications only by their keywords [note that it will only work with keywords](and some filters [see kwargs field])
    Arguments available: keywords: string [mandatory, separated by space], limit [mandatory, default = 5]
    Kwargs available [optional]: (release: integer as string or 'Rel-xxx', wg: string = working group (S1, C4, SP, ...), spec_type: string (either TS or TR), mode: string (either 'and' or 'or') = search mode)
    """
    endpoint = DOC3GPPFINDER_BASE + "/search-spec"
    data = await post_data_to_api(endpoint, keywords=keywords)
    if isinstance(data, tuple) and data[0] is None:
        return f"An error has occured while searching specifications"
    results = data['results'][:min(len(data['results'])-1, limit)]
    output = []
    for spec in results:
        output.append(f"Specification ID: {spec['id']}\nTitle: {spec['title']}\nType: {spec['type']}\nRelease: {spec['release']}\nVersion: {spec['version']}\nWorking Group: {spec['working_group']}\nURL of spec: {spec['url']}\n")
    
    return "-\n".join(output)

@mcp.tool()
async def get_multiple_documents_url(doc_ids: List[str], release: int = None):
    """
    [BATCH] Search multiple 3GPP documents (TSG docs, specifications or workshop files) [note that it will only work with document ID] (and release if it's a specification only) and return only their position via a URL
    Arguments available: doc_ids: list of string [mandatory], release: integer [optional for every case]
    """
    endpoint = DOC3GPPFINDER_BASE + "/batch"
    data = await post_data_to_api(endpoint, doc_ids=doc_ids, release=release)
    if isinstance(data, tuple) and data[0] is None:
        return f"An error while searching publications: {data[1]}"
    results = data["results"]
    output = []
    for doc_id, url in results.items():
        output.append(f'Document ID: {doc_id}\nURL: {url}\n')
    return "-\n".join(output)

# PTT Endpoints

@mcp.tool()
async def search_documents_web(query: str, data_type: str = None, limit: int = 5):
    """
    Search on the Web (thanks to DuckDuckGo) documents based on the user's query
    Arguments available: query: string [mandatory], data_type: string [optional, either 'pdf', 'patent' or None (classic web search)], limit: integer [optional, default = 5]
    """
    endpoint = DUCKDUCKGO_BASE + "/search"
    data = await fake_post_data_to_api(endpoint, query=query, data_type=data_type, max_references=limit)
    if isinstance(data, tuple) and data[0] is None:
        return f"An error while searching publications: {data[1]}"
    results = data["results"]
    output = []
    for ref in results:
        output.append(f"Title: {ref['title']}\nBody: {ref['body']}\nURL: {ref['url']}")
    return "-\n".join(output)

if __name__ == "__main__":
    mcp.run(transport="stdio")