Spaces:
Running
Running
File size: 4,026 Bytes
29a7606 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
"""
MCP Web Server using FastMCP
"""
import sys
import os
# Add src to pythonpath so imports work
current_dir = os.path.dirname(os.path.abspath(__file__))
src_dir = os.path.dirname(os.path.dirname(current_dir))
if src_dir not in sys.path:
sys.path.append(src_dir)
from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Any, Union
from core.mcp_telemetry import log_usage, log_trace, log_metric
import uuid
import time
# Local imports
try:
from .tools.search import search_web
from .tools.extract import extract_content
from .tools.research import research_topic
from .tools.wikipedia import search_wikipedia, get_wikipedia_page
from .tools.arxiv import search_arxiv
except ImportError:
# Fallback if run directly
try:
from tools.search import search_web
from tools.extract import extract_content
from tools.research import research_topic
from tools.wikipedia import search_wikipedia, get_wikipedia_page
from tools.arxiv import search_arxiv
except ImportError:
# Fallback if tools are relative to this file but not package
sys.path.append(os.path.join(current_dir, "tools"))
from search import search_web
from extract import extract_content
from research import research_topic
from wikipedia import search_wikipedia, get_wikipedia_page
from arxiv import search_arxiv
# Initialize FastMCP Server
mcp = FastMCP("MCP Web", host="0.0.0.0")
@mcp.tool()
def search(query: str, max_results: int = 5) -> List[Dict[str, Any]]:
"""
Search the web for the given query using DuckDuckGo.
Returns a list of results with title, url, snippet.
"""
start_time = time.time()
trace_id = str(uuid.uuid4())
span_id = str(uuid.uuid4())
log_usage("mcp-web", "search")
try:
results = search_web(query, max_results)
duration = (time.time() - start_time) * 1000
log_trace("mcp-web", trace_id, span_id, "search", duration, "ok")
log_metric("mcp-web", "search_results_count", len(results), {"query": query})
return results
except Exception as e:
duration = (time.time() - start_time) * 1000
log_trace("mcp-web", trace_id, span_id, "search", duration, "error")
raise e
@mcp.tool()
def extract(url: str) -> str:
"""
Extracts text content from a given URL.
Useful for reading articles or documentation.
"""
log_usage("mcp-web", "extract")
return extract_content(url)
@mcp.tool()
def research(query: str, max_results: int = 3) -> List[Dict[str, Any]]:
"""
Research a topic by searching and extracting content in parallel.
Returns search results populated with full content.
"""
log_usage("mcp-web", "research")
return research_topic(query, max_results)
@mcp.tool()
def wikipedia_search(query: str, max_results: int = 5) -> List[str]:
"""
Search Wikipedia for the given query.
Returns a list of page titles.
"""
log_usage("mcp-web", "wikipedia_search")
return search_wikipedia(query, max_results)
@mcp.tool()
def wikipedia_page(title: str) -> Dict[str, Any]:
"""
Get the content of a Wikipedia page.
Returns title, content, summary, url.
"""
log_usage("mcp-web", "wikipedia_page")
return get_wikipedia_page(title)
@mcp.tool()
def arxiv_search(query: str, max_results: int = 5) -> List[Dict[str, Any]]:
"""
Search Arxiv for papers.
Returns metadata including title, summary, authors, pdf_url.
"""
log_usage("mcp-web", "arxiv_search")
return search_arxiv(query, max_results)
if __name__ == "__main__":
# Run the MCP server
import os
if os.environ.get("MCP_TRANSPORT") == "sse":
import uvicorn
port = int(os.environ.get("PORT", 7860))
uvicorn.run(mcp.sse_app(), host="0.0.0.0", port=port)
else:
mcp.run()
|