File size: 3,307 Bytes
8bab08d |
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 |
# file: mcp/servers/search_server.py
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
from datetime import datetime
from aiohttp import web
import logging
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from services.web_search import get_search_service
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SearchServer:
"""Real search MCP server using Serper API (serper.dev)"""
def __init__(self):
self.search_service = get_search_service()
logger.info("Search MCP Server initialized with Serper API (serper.dev)")
async def handle_rpc(self, request):
data = await request.json()
method = data.get("method")
params = data.get("params", {})
if method == "health":
return web.json_response({"result": "ok"})
elif method == "search.query":
q = params.get("q", "")
max_results = params.get("max_results", 5)
if not q:
return web.json_response({"error": "Query parameter 'q' is required"}, status=400)
logger.info(f"Search query: '{q}'")
# Perform real web search
search_results = await self.search_service.search(q, max_results=max_results)
# Format results for MCP protocol
results = []
for result in search_results:
results.append({
"text": result.get('body', ''),
"title": result.get('title', ''),
"source": result.get('source', ''),
"url": result.get('url', ''),
"ts": datetime.utcnow().isoformat(),
"confidence": 0.8 # Base confidence for real search results
})
logger.info(f"Returning {len(results)} search results")
return web.json_response({"result": results})
elif method == "search.news":
q = params.get("q", "")
max_results = params.get("max_results", 5)
if not q:
return web.json_response({"error": "Query parameter 'q' is required"}, status=400)
logger.info(f"News search query: '{q}'")
# Perform news search
news_results = await self.search_service.search_news(q, max_results=max_results)
# Format results
results = []
for result in news_results:
results.append({
"text": result.get('body', ''),
"title": result.get('title', ''),
"source": result.get('source', ''),
"url": result.get('url', ''),
"date": result.get('date', ''),
"ts": datetime.utcnow().isoformat(),
"confidence": 0.85 # Higher confidence for news
})
logger.info(f"Returning {len(results)} news results")
return web.json_response({"result": results})
return web.json_response({"error": f"Unknown method: {method}"}, status=400)
app = web.Application()
server = SearchServer()
app.router.add_post("/rpc", server.handle_rpc)
if __name__ == "__main__":
web.run_app(app, port=9001) |