#!/usr/bin/env python3 """Deduped MCP server: AI use cases from hot AI repos, the hot repos and today's AI news, for any MCP client. One file, Python 3.9+ standard library only. Speaks MCP over stdio (newline-delimited JSON-RPC 2.0) and only READS the public static JSON at https://dedupedai.com/api/v1/ over HTTPS. No key, no account, nothing is written anywhere. Tools: search_use_cases, get_use_case, hot_repos, today_news. Env: DEDUPED_API_BASE (default https://dedupedai.com/api/v1), DEDUPED_CACHE_S (default 600 seconds). """ import json, os, re, sys, time, urllib.request, urllib.error __version__ = "0.1.0" API = os.environ.get("DEDUPED_API_BASE", "https://dedupedai.com/api/v1").rstrip("/") CACHE_S = int(os.environ.get("DEDUPED_CACHE_S") or 600) PROTOCOL = "2025-06-18" UA = f"deduped-mcp/{__version__} (+https://dedupedai.com/api/)" ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*--[a-z0-9][a-z0-9-]*$") _cache = {} def fetch(path, opener=None): """GET API/path as JSON, cached for CACHE_S seconds. opener is injectable for tests.""" now = time.time() hit = _cache.get(path) if hit and now - hit[0] < CACHE_S: return hit[1] req = urllib.request.Request(f"{API}/{path}", headers={"User-Agent": UA, "Accept": "application/json"}) with (opener or urllib.request.urlopen)(req, timeout=20) as r: doc = json.loads(r.read().decode("utf-8")) _cache[path] = (now, doc) return doc def _words(s): return re.findall(r"[a-z0-9]+", (s or "").lower()) def search_use_cases(query, category=None, limit=8, opener=None): idx = fetch("search.json", opener).get("items", []) q = set(_words(query)) - {"a", "an", "the", "to", "for", "with", "and", "or", "of", "in", "my", "i", "how", "use"} out = [] for it in idx: if category and category.lower() not in (it.get("c") or "").lower(): continue fields = [(it.get("t"), 3.0), (it.get("rt"), 2.5), (it.get("r"), 2.0), (it.get("s"), 1.5), (it.get("o"), 1.0), (it.get("w"), 1.0), (it.get("c"), 0.5)] score = 0.0 for text, wgt in fields: ws = set(_words(text)) score += wgt * sum(1 for t in q if t in ws or any(w.startswith(t) for w in ws if len(t) > 3)) if q and score == 0: continue score += 0.5 * (it.get("sc") or 0) + (0.3 if it.get("b") else 0) + (0.2 if it.get("f") else 0) out.append((score, it)) out.sort(key=lambda x: -x[0]) base = API.rsplit("/api/", 1)[0] return [{"id": it["id"], "title": it.get("t"), "summary": it.get("s"), "repo": it.get("r"), "category": it.get("c"), "best": it.get("b"), "seen_in_the_wild": it.get("f"), "url": base + it.get("u", "")} for _, it in out[:max(1, min(int(limit or 8), 25))]] def get_use_case(id, opener=None): if not ID_RE.match(id or ""): raise ValueError("id must look like repo-slug--use-case-slug (from search_use_cases)") doc = fetch(f"usecases/{id}.json", opener) keep = ("id", "title", "repo", "repo_url", "summary", "who", "problem", "outcome", "steps", "agent_prompt", "requirements", "risks", "caveat", "difficulty", "time", "agents", "seen_in_the_wild", "url", "disclaimer") return {k: doc.get(k) for k in keep if k in doc} def hot_repos(limit=10, opener=None): items = fetch("repos.json", opener).get("items", []) items = [x for x in items if x.get("show", True)] return [{k: x.get(k) for k in ("repo", "title", "what", "language", "stars", "gain", "url", "use_cases", "uc_n")} for x in items[:max(1, min(int(limit or 10), 50))]] def today_news(opener=None): d = fetch("news.json", opener) return {"date": d.get("date"), "subject": d.get("subject"), "url": d.get("url"), "items": [{k: x.get(k) for k in ("rank", "title", "category", "coverage_tag", "body", "story_url")} for x in d.get("items", [])]} TOOLS = [ {"name": "search_use_cases", "description": "Search Deduped's catalog of practical AI use cases generated from hot GitHub AI repos. " "Returns ids, titles, summaries and page urls. Then call get_use_case for steps and an agent prompt.", "inputSchema": {"type": "object", "required": ["query"], "properties": { "query": {"type": "string", "description": "The task, e.g. 'turn PDFs into a searchable knowledge base'"}, "category": {"type": "string", "description": "Optional category filter, e.g. Research, Content, Software and DevOps"}, "limit": {"type": "integer", "minimum": 1, "maximum": 25, "default": 8}}}, "annotations": {"readOnlyHint": True, "openWorldHint": True}}, {"name": "get_use_case", "description": "One use case: who it is for, steps, a copy-paste agent prompt, requirements, risks, caveat and the repo url.", "inputSchema": {"type": "object", "required": ["id"], "properties": { "id": {"type": "string", "description": "Use case id from search_use_cases, e.g. owner-repo--use-case-slug"}}}, "annotations": {"readOnlyHint": True, "openWorldHint": True}}, {"name": "hot_repos", "description": "The hottest AI repos on GitHub right now (last 90 days, best first), with a plain-English line each.", "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 10}}}, "annotations": {"readOnlyHint": True, "openWorldHint": True}}, {"name": "today_news", "description": "Today's deduped AI news brief: the top stories, each once, with category and a short summary.", "inputSchema": {"type": "object", "properties": {}}, "annotations": {"readOnlyHint": True, "openWorldHint": True}}, ] CALLS = {"search_use_cases": search_use_cases, "get_use_case": get_use_case, "hot_repos": hot_repos, "today_news": today_news} def handle(msg, opener=None): """One JSON-RPC message in, one response dict out (None for notifications).""" mid, method, params = msg.get("id"), msg.get("method"), msg.get("params") or {} if mid is None: return None ok = lambda result: {"jsonrpc": "2.0", "id": mid, "result": result} if method == "initialize": return ok({"protocolVersion": params.get("protocolVersion") or PROTOCOL, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "deduped", "title": "Deduped AI use cases", "version": __version__}, "instructions": "Use search_use_cases to find a practical AI use case for the user's task, then " "get_use_case for steps and an agent prompt. Link the use case url when you show it."}) if method == "ping": return ok({}) if method == "tools/list": return ok({"tools": TOOLS}) if method == "tools/call": name, args = params.get("name"), params.get("arguments") or {} fn = CALLS.get(name) if not fn: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32602, "message": f"unknown tool {name}"}} try: data = fn(**args, opener=opener) return ok({"content": [{"type": "text", "text": json.dumps(data, ensure_ascii=False, indent=1)}], "structuredContent": data if isinstance(data, dict) else {"items": data}, "isError": False}) except urllib.error.HTTPError as ex: text = "Not found. Use search_use_cases to get a valid id." if ex.code == 404 else f"HTTP {ex.code} from {API}" except (TypeError, ValueError, OSError) as ex: text = f"{type(ex).__name__}: {ex}" return ok({"content": [{"type": "text", "text": text}], "isError": True}) return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": f"method not found: {method}"}} def main(stdin=None, stdout=None, opener=None): stdin, stdout = stdin or sys.stdin, stdout or sys.stdout for line in stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except ValueError: resp = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "parse error"}} else: resp = handle(msg, opener) if isinstance(msg, dict) else None if resp is not None: stdout.write(json.dumps(resp, ensure_ascii=False) + "\n") stdout.flush() if __name__ == "__main__": main()