OrganizedProgrammers/SHTTPMCPServer

1from typing import *2import httpx3from mcp.server.fastmcp import FastMCP4 5server = FastMCP(name="streamable-http-mcp-server-test", json_response=False, stateless_http=False)6 7async def make_request(url: str, method: Literal["GET", "POST"], data: Dict[str, Any] = {}):8    headers = {"Accept": "application/json"}9    async with httpx.AsyncClient(verify=False) as client:10        try:11            if method == "GET":12                response = await client.get(url, headers=headers)13            elif method == "POST":14                response = await client.post(url, headers=headers, json=data)15            else:16                print("Method not allowed !")17                return None18            response.raise_for_status()19            return response.json()20        except:21            return None22 23# arXiv24@server.tool()25async def search_academic_papers_arxiv(keyword: str, limit: int = 5) -> str:26    """27    Search papers from arXiv database with specified keywords [optional: a limit of papers the user wants]28    Args: keyword: string, [optional: limit: integer, set limit to 5 if not specified]29    """30    response = await make_request("https://organizedprogrammers-arxiv.hf.space/search", "POST", {"keyword": keyword, "limit": limit})31    if not response:32        return "Unable to find papers | No papers has been found"33    return "\n".join([f"arXiv n°{paper_id} - {paper_meta['title']} by {paper_meta['authors']} : {paper_meta['abstract']}" for paper_id, paper_meta in response['message'].items()])34 35@server.tool()36async def get_arxiv_pub_text(arxiv_id: str) -> str:37    """38    Extract publication PDF via arXiv ID39    Returns the full content of the publication40    Args: arxiv_id -> string41    """42    response = await make_request("https://organizedprogrammers-arxiv.hf.space/extract_pdf/arxiv_id", "POST", {"doc_id": arxiv_id})43    if not response:44        return "Unable to extract PDF | arXiv PDF not found"45    return response["message"]["text"]46 47# DocFinder48@server.tool()49async def get_document_url(doc_id: str) -> str:50    """51    Find technical document or specification from 3GPP / ETSI / GP by a document ID52    Returns the URL (also scope + version if doc is a specification [not all specifications have a version or scope])53    Arguments: doc_id -> string54    """55    response = await make_request('https://organizedprogrammers-docfinder.hf.space/find/single', "POST", {"doc_id": doc_id})56    if not response:57        return "Unable to find document/specification"58    version = response.get('version', 'unavailable')59    scope = response.get('scope', 'unavailable')60    return f'Downloadable !\nDoc No. {doc_id}\nURL : {response.get("url")}\nVersion : {version}\nScope : {scope}'61 62@server.tool()63async def search_specifications_with_keywords(keywords: str, threshold: int = 60, source: Literal["3GPP", "ETSI", "all"] = "all", spec_type: Optional[Literal["TS", "TR"]] = None):64    """65    Search specifications from 3GPP and/or ETSI with keywoeds (Based off BM25 scoring)66    Returns a list of specifications metadata that matches the similarity score threshold, the keywords, the source and specification type67    Arguments:68        - keywords -> string69        - threshold -> integer (by default, set to 60) [between 0-100]70        - source -> string (either '3GPP', 'ETSI' or 'all', by default, set to 'all')71        - spec_type -> string (either 'TS' or 'TR' or None, by default, set to None)72    """73    response = await make_request('https://organizedprogrammers-docfinder.hf.space/search/bm25', "POST", {"keywords": keywords, "threshold": threshold, "source": source, "spec_type": spec_type})74    if not response:75        return "Unable to search specifications | No specifications has been found"76    results = response["results"]77    return "\n---\n".join([f"Specification ID: {spec['id']}\nTitle: {spec['title']}\nType: {'Technical Specification' if spec['spec_type'] == 'TS' else 'Technical Report'}\nVersion: {spec.get('version', 'unavailable')}\nScope: {spec.get('scope', 'unavailable')}\nWorking Group: {spec.get('working_group', 'not defined')}\nURL: {spec.get('url', 'unavailable')}" for spec in results])78 79# SpecSplitter80@server.tool()81async def get_spec_text(spec_id: str) -> str:82    """83    Extract specification from 3GPP or ETSI84    Returns a dictionary k:v where k is the section (1., 2.2.1, ...) and v, the content of k, or a string if failed85    Args: spec_id -> string86    """87    response = await make_request('https://organizedprogrammers-specsplitter.hf.space/extract_text/structured', "POST", {"spec_id": spec_id})88    if not response:89        return "Unable to extract specification text"90    return "\n".join([f"{k}: {v}" for k, v in response.keys()])91 92# SERPent93 94@server.tool()95async def search_google_patents(queries: List[str], n_results: int) -> str:96    """97    Search patents from Google Patents98    You can generate multiple queries (at least 1)99    Returns a list of patents from queries, for each query, {n_results} patents will be retrieved100    Args: queries -> list of string, n_results -> integer [by default: 10]101    """102    response = await make_request("https://organizedprogrammers-serpent.hf.space/serp/search_patents", "POST", {"queries": queries, "n_results": n_results})103    if not response:104        return "Unable to fetch patents"105    return "\n".join(f"[Patent ID: {patent['id']} | Title: {patent['title']} | Body: {patent['body']}]" for patent in response.results)106 107@server.tool()108async def scrap_google_patents(patent_ids: List[str]) -> str:109    """110    Scrap patents from one or many patents from Google Patents111    Returns a list of patents with their title, abstract, description, claims, field of invention and background112    Args: patent_ids -> list of strings corresponding to Google Patent ID [min. 1]113    """114    if len(patent_ids) > 1:115        response = await make_request("https://organizedprogrammers-serpent.hf.space/scrap/scrap_patents_bulk", "POST", {"patent_ids": patent_ids})116        if not response:117            return "Unable to scrap patents"118        return "\n---\n".join([f"Title: {pat['title']}\nAbstract: {pat['abstract']}\nDescription: {pat['description']}\nClaims: {pat['claims']}\nField of invention{pat['field_of_invention']}\nBackground: {pat['background']}" for pat in response['patents']])119    elif len(patent_ids) == 1:120        response = await make_request("https://organizedprogrammers-serpent.hf.space/scrap/scrap_patent/"+patent_ids[0], "GET")121        if not response:122            return "Unable to scrap patent"123        return f"Title: {response['title']}\nAbstract: {response['abstract']}\nDescription: {response['description']}\nClaims: {response['claims']}\nField of invention{response['field_of_invention']}\nBackground: {response['background']}"124 125app = server.streamable_http_app