connor
b4bc5e1031
searxng-mcp.py is a stdlib-only stdio MCP server with one web_search tool against search.rcjohnstone.com. It is in both profiles and asks first on oc-work. lsp: true in opencode.json, and the launcher sets OPENCODE_EXPERIMENTAL_LSP_TOOL and OPENCODE_DISABLE_LSP_DOWNLOAD. The launcher adds gitea-mcp to oc-local through OPENCODE_CONFIG_CONTENT when the binary and GITEA_MCP_TOKEN exist, limited to 15 tools. gitea_issue_write asks.
129 lines
4.3 KiB
Python
Executable File
129 lines
4.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""MCP server (stdio) exposing one tool, web_search, backed by a SearXNG instance.
|
|
|
|
Standard library only. Set SEARXNG_URL to override the instance.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
SEARXNG_URL = os.environ.get("SEARXNG_URL", "https://search.rcjohnstone.com").rstrip("/")
|
|
MAX_RESULTS = 10
|
|
TIMEOUT_S = 20
|
|
KNOWN_VERSIONS = ["2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"]
|
|
|
|
TOOL = {
|
|
"name": "web_search",
|
|
"description": (
|
|
"Search the web through Connor's self-hosted SearXNG. Returns up to "
|
|
f"{MAX_RESULTS} results with title, URL and snippet. Use webfetch on a "
|
|
"result URL to read the page."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "Search terms."},
|
|
"page": {"type": "integer", "minimum": 1, "default": 1},
|
|
"time_range": {
|
|
"type": "string",
|
|
"enum": ["day", "month", "year"],
|
|
"description": "Only results from this period.",
|
|
},
|
|
"category": {
|
|
"type": "string",
|
|
"enum": ["general", "it", "news", "science"],
|
|
"default": "general",
|
|
},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
}
|
|
|
|
|
|
def search(args):
|
|
query = str(args.get("query", "")).strip()
|
|
if not query:
|
|
raise ValueError("query is empty")
|
|
params = {
|
|
"q": query,
|
|
"format": "json",
|
|
"pageno": int(args.get("page", 1)),
|
|
"categories": args.get("category", "general"),
|
|
}
|
|
if args.get("time_range"):
|
|
params["time_range"] = args["time_range"]
|
|
url = f"{SEARXNG_URL}/search?{urllib.parse.urlencode(params)}"
|
|
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=TIMEOUT_S) as resp:
|
|
data = json.load(resp)
|
|
|
|
lines = []
|
|
for answer in data.get("answers", [])[:2]:
|
|
text = answer.get("answer") if isinstance(answer, dict) else answer
|
|
if text:
|
|
lines.append(f"Answer: {text}\n")
|
|
for i, r in enumerate(data.get("results", [])[:MAX_RESULTS], 1):
|
|
snippet = " ".join((r.get("content") or "").split())
|
|
lines.append(f"{i}. {r.get('title', '').strip()}\n {r.get('url', '')}\n {snippet}")
|
|
if not lines:
|
|
lines.append("No results.")
|
|
if data.get("suggestions"):
|
|
lines.append("Suggestions: " + ", ".join(data["suggestions"][:5]))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def handle(msg):
|
|
method = msg.get("method")
|
|
if method == "initialize":
|
|
requested = msg.get("params", {}).get("protocolVersion")
|
|
version = requested if requested in KNOWN_VERSIONS else KNOWN_VERSIONS[-2]
|
|
return {
|
|
"protocolVersion": version,
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": {"name": "searxng", "version": "1.0.0"},
|
|
}
|
|
if method == "ping":
|
|
return {}
|
|
if method == "tools/list":
|
|
return {"tools": [TOOL]}
|
|
if method == "tools/call":
|
|
params = msg.get("params", {})
|
|
if params.get("name") != TOOL["name"]:
|
|
raise LookupError(f"unknown tool: {params.get('name')}")
|
|
try:
|
|
text, is_error = search(params.get("arguments") or {}), False
|
|
except (urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as e:
|
|
text, is_error = f"Search failed: {e}", True
|
|
return {"content": [{"type": "text", "text": text}], "isError": is_error}
|
|
raise LookupError(f"method not found: {method}")
|
|
|
|
|
|
def main():
|
|
for line in sys.stdin:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
msg = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if "id" not in msg:
|
|
continue
|
|
reply = {"jsonrpc": "2.0", "id": msg["id"]}
|
|
try:
|
|
reply["result"] = handle(msg)
|
|
except LookupError as e:
|
|
reply["error"] = {"code": -32601, "message": str(e)}
|
|
except Exception as e:
|
|
reply["error"] = {"code": -32603, "message": str(e)}
|
|
sys.stdout.write(json.dumps(reply) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|