Spaces:
Sleeping
Sleeping
| from typing import * | |
| import httpx | |
| from mcp.server.fastmcp import FastMCP | |
| server = FastMCP(name="streamable-http-mcp-server-test", json_response=False, stateless_http=False) | |
| async def make_request(url: str, data: Dict[str, Any]): | |
| headers = {"Accept": "application/json"} | |
| async with httpx.AsyncClient(verify=False) as client: | |
| try: | |
| response = await client.post(url, headers=headers, json=data) | |
| response.raise_for_status() | |
| return response.json() | |
| except: | |
| return None | |
| async def search_academic_papers_arxiv(keyword: str, limit: int = 5) -> str: | |
| """ | |
| Search papers from arXiv database with specified keywords [optional: a limit of papers the user wants] | |
| Args: keyword: string, [optional: limit: integer, set limit to 5 if not specified] | |
| """ | |
| response = await make_request("https://organizedprogrammers-arxiv.hf.space/search", {"keyword": keyword, "limit": limit}) | |
| if not response: | |
| return "Unable to find papers | No papers has been found" | |
| 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()]) | |
| async def get_arxiv_pub_text(arxiv_id: str) -> str: | |
| """ | |
| Extract publication PDF via arXiv ID | |
| Returns the full content of the publication | |
| Args: arxiv_id -> string | |
| """ | |
| response = await make_request("https://organizedprogrammers-arxiv.hf.space/extract_pdf/arxiv_id", {"doc_id": arxiv_id}) | |
| if not response: | |
| return "Unable to extract PDF | arXiv PDF not found" | |
| return response["message"]["text"] | |
| async def get_document_url(doc_id: str) -> str: | |
| """ | |
| Find technical document or specification from 3GPP / ETSI / GP by a document ID | |
| Returns the URL (also scope + version if doc is a specification [not all specifications have a version or scope]) | |
| Arguments: doc_id -> string | |
| """ | |
| response = await make_request('https://organizedprogrammers-docfinder.hf.space/find/single', {"doc_id": doc_id}) | |
| if not response: | |
| return "Unable to find document/specification" | |
| version = response.get('version', 'unavailable') | |
| scope = response.get('scope', 'unavailable') | |
| return f'Downloadable !\nDoc No. {doc_id}\nURL : {response.get("url")}\nVersion : {version}\nScope : {scope}' | |
| async def search_specifications_with_keywords(keywords: str, threshold: int = 60, source: Literal["3GPP", "ETSI", "all"] = "all", spec_type: Optional[Literal["TS", "TR"]] = None): | |
| """ | |
| Search specifications from 3GPP and/or ETSI with keywoeds (Based off BM25 scoring) | |
| Returns a list of specifications metadata that matches the similarity score threshold, the keywords, the source and specification type | |
| Arguments: | |
| - keywords -> string | |
| - threshold -> integer (by default, set to 60) [between 0-100] | |
| - source -> string (either '3GPP', 'ETSI' or 'all', by default, set to 'all') | |
| - spec_type -> string (either 'TS' or 'TR' or None, by default, set to None) | |
| """ | |
| response = await make_request('https://organizedprogrammers-docfinder.hf.space/search/bm25', {"keywords": keywords, "threshold": threshold, "source": source, "spec_type": spec_type}) | |
| if not response: | |
| return "Unable to search specifications | No specifications has been found" | |
| results = response["results"] | |
| 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]) | |
| async def get_spec_text(spec_id: str) -> Union[Dict[str, str], str]: | |
| """ | |
| Extract specification from 3GPP or ETSI | |
| Returns a dictionary k:v where k is the section (1., 2.2.1, ...) and v, the content of k, or a string if failed | |
| Args: spec_id -> string | |
| """ | |
| response = await make_request('https://organizedprogrammers-specsplitter.hf.space/extract_text/structured', {"spec_id": spec_id}) | |
| if not response: | |
| return "Unable to extract specification text" | |
| return response | |
| app = server.streamable_http_app |