#!/usr/bin/env python3 """Guaq AI jobs — an MCP server, so an LLM can query the index directly. Run it from any MCP host (Claude Desktop, Claude Code, an agent framework): { "mcpServers": { "guaq": { "command": "python", "args": ["/path/to/guaq_mcp.py"], "env": {"GUAQ_API_KEY": "guaq_sk_..."} } } } WHY THIS IS HAND-WRITTEN JSON-RPC --------------------------------- The MCP wire protocol over stdio is JSON-RPC 2.0 with three methods that matter (initialize, tools/list, tools/call). Implementing them directly is about a hundred lines and costs the user no `pip install` — which matters because this file runs on the customer's machine, not ours, often inside a locked-down Python where adding a dependency is a ticket. If this ever needs resources, prompts or sampling, take the SDK then. TWO THINGS IT IS CAREFUL ABOUT ------------------------------ 1. RESULTS ARE TRIMMED FOR A CONTEXT WINDOW, NOT FOR A DATAFRAME. A tool call returns at most 50 rows and a short field set. An LLM asked to summarise a market does not need the apply URL of every one of 40,000 postings, and filling its context with them makes the answer worse, not better. Bulk pulls go through export_jobs, which writes a file. 2. AN ERROR COMES BACK AS TEXT THE MODEL CAN ACT ON. A quota refusal or an upgrade requirement is reported in words ("your plan does not include market signals"), not as a transport-level failure, so the model tells the user what to do instead of retrying a call that will never succeed. """ from __future__ import annotations import json import os import sys import traceback from guaq_client import Guaq, GuaqError PROTOCOL_VERSION = "2024-11-05" SERVER_INFO = {"name": "guaq-jobs", "version": "1.0.0"} # The exact values the API stores. Hardcoded, not imported from taxonomy/sector_adapters: # this file is downloaded standalone from /v1/mcp and runs beside guaq_client.py only. # tests/test_guaq_mcp.py fails if these drift from the server's own vocabularies. They # matter more here than anywhere: an LLM copies the description literally, and one wrong # spelling ("full-time" for full_time) is a silent zero-row answer, not an error. DOMAINS = ["ai_ml", "business", "consulting", "data", "design_media", "devops_cloud", "embedded", "finance", "hardware_engineering", "hr_people", "law_policy", "life_sciences", "marketing", "operations", "product", "research", "security", "tech", "unspecified"] EMPLOYMENT_TYPES = ["internship", "apprenticeship", "trainee", "new_grad_program", "campus_hire", "full_time", "part_time", "contract", "permanent", "fellowship", "postdoc", "freelance", "volunteer"] ROLE_TYPES = ["internship", "new_grad", "junior", "mid", "senior", "lead_exec", "founding_engineer"] WORK_MODES = ["remote", "hybrid", "on_site", "unspecified"] VISA = ["stated", "likely", "refused"] _FILTERS = { "q": "Text to match in the job title or company name.", "company": "Employer name, case-insensitive. Comma-separated for several.", "domain": "Industry domain (exact id).", "category": "Taxonomy path prefix, e.g. 'engineering/software' — matches that node and everything under it.", "role_type": "Seniority tier.", "employment_type": "Employment type — the exact id, with underscores (full_time, part_time).", "location_scope": "Country bucket, lowercase: india, us, uk, sg, or a lowercase ISO " "country code for anywhere else (de, ca, jp, ...); also remote, " "multiple, unspecified. India is 'india' (not 'in') and the UK is " "'uk' (not 'gb'). There is no 'eu' bucket.", "city": "City name. Matched as a substring with synonyms, so 'bangalore' also finds 'Bengaluru'.", "work_mode": "Work arrangement.", "tech": "A technology named in the stack, e.g. kubernetes. Comma-separated for several.", "visa": "stated (the posting says it sponsors), likely (stated or inferred), or refused.", "yoe_min": "Minimum years of experience.", "yoe_max": "Maximum years of experience.", "grad_year": "Graduating cohort the role targets.", "seen_after": "ISO date. Only roles confirmed live on or after this date.", } _ENUMS = {"domain": DOMAINS, "employment_type": EMPLOYMENT_TYPES, "role_type": ROLE_TYPES, "work_mode": WORK_MODES, "visa": VISA} _FILTER_SCHEMA = {k: dict({"type": "string", "description": v}, **({"enum": _ENUMS[k]} if k in _ENUMS else {})) for k, v in _FILTERS.items()} TOOLS = [ { "name": "search_jobs", "description": "Search live job postings across ~5,000 employers. Returns at most " "50 rows plus has_more/next_cursor — narrow with filters rather " "than paging far, and call count_jobs if you need the size of " "the match.", "inputSchema": { "type": "object", "properties": dict(_FILTER_SCHEMA, limit={"type": "integer", "description": "Rows to return, 1-50. Default 25."}, cursor={"type": "integer", "description": "next_cursor from a previous " "search_jobs call, for the next " "page."}), }, }, { "name": "count_jobs", "description": "How many postings match a set of filters. Cheap, and the right " "first call when the question is about volume or share rather " "than about specific roles.", "inputSchema": {"type": "object", "properties": dict(_FILTER_SCHEMA)}, }, { "name": "get_job", "description": "One posting in full, by id.", "inputSchema": {"type": "object", "properties": {"id": {"type": "integer"}}, "required": ["id"]}, }, { "name": "list_companies", "description": "Employers in the index with their live role counts. Accepts the " "same filters as search_jobs, so it answers 'who is hiring " "quant developers in Bengaluru' directly.", "inputSchema": { "type": "object", "properties": dict(_FILTER_SCHEMA, limit={"type": "integer", "description": "Default 50."}), }, }, { "name": "company_profile", "description": "One employer's hiring profile: role mix, seniority mix, " "geographies, technologies, and the flow of postings over recent " "weeks.", "inputSchema": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}, }, { "name": "market_signal", "description": "Aggregate market intelligence derived from the whole index. " "Requires an organisation seat. Signals: overview, technologies, " "vendors, flow, india, gcc, quant, it_services, semis, baskets, " "sector, function_mix.", "inputSchema": { "type": "object", "properties": { "signal": {"type": "string", "description": "Which signal to fetch."}, "company": {"type": "string"}, "city": {"type": "string"}, "category": {"type": "string"}, "ticker": {"type": "string"}, "tickers": {"type": "string", "description": "Comma-separated."}, "weeks": {"type": "integer"}, }, "required": ["signal"], }, }, { "name": "export_jobs", "description": "Write every matching posting to a local file (jsonl, csv or " "parquet) for analysis. Use this instead of paging search_jobs " "when the answer needs the whole dataset. Requires Ultimate or an " "organisation seat.", "inputSchema": { "type": "object", "properties": dict( _FILTER_SCHEMA, path={"type": "string", "description": "Where to write the file."}, format={"type": "string", "enum": ["jsonl", "csv", "parquet"]}), "required": ["path"], }, }, { "name": "api_meta", "description": "The caller's plan, remaining quota, and the exact vocabularies " "the filters accept (domains, role types, cities). Worth calling " "once before guessing a filter value.", "inputSchema": {"type": "object", "properties": {}}, }, ] # Field sets kept deliberately narrow — see note 1 in the module docstring. _ROW_FIELDS = ("id", "company", "title", "location", "location_scope", "role_type", "seniority_label", "employment_type", "yoe_min", "work_arrangement", "category_path", "tech_stack", "posted_at", "url") def _trim(row: dict) -> dict: return {k: row.get(k) for k in _ROW_FIELDS if row.get(k) is not None} def _filters(args: dict) -> dict: return {k: v for k, v in args.items() if k in _FILTERS and v not in (None, "")} def call_tool(g: Guaq, name: str, args: dict): if name == "search_jobs": # No count here: it would double the caller's quota and run the expensive COUNT # on every search. count_jobs exists for when the size is the question. limit = max(1, min(int(args.get("limit") or 25), 50)) page = g.jobs(limit=limit, cursor=args.get("cursor"), **_filters(args)) rows = page["data"] return {"showing": len(rows), "has_more": bool(page.get("has_more")), "next_cursor": page.get("next_cursor"), "jobs": [_trim(r) for r in rows], "note": ("More rows match. Narrow the filters, pass next_cursor for the " "next page, or use export_jobs for the full set." if page.get("has_more") else None)} if name == "count_jobs": return {"count": g.count(**_filters(args))} if name == "get_job": raw = args.get("id") try: job_id = int(raw) except (TypeError, ValueError): raise GuaqError(f"id must be an integer job id, got {raw!r}", 400) return g.job(job_id) if name == "list_companies": rows = g.companies(limit=max(1, min(int(args.get("limit") or 50), 200)), **_filters(args)) return {"companies": rows} if name == "company_profile": return g.company(args["name"]) if name == "market_signal": params = {k: v for k, v in args.items() if k in ("company", "city", "category", "ticker", "tickers", "weeks") and v} return g.signal(args["signal"], **params) if name == "export_jobs": path = args["path"] n = g.export(path, fmt=args.get("format") or "jsonl", **_filters(args)) return {"written": path, "bytes": n, "note": "Read it with pandas/polars; it is the complete match, not a page."} if name == "api_meta": return g.meta() raise GuaqError(f"Unknown tool: {name}", 404) def _result(payload) -> dict: return {"content": [{"type": "text", "text": json.dumps(payload, indent=2, default=str)}]} def _error(text: str) -> dict: return {"content": [{"type": "text", "text": text}], "isError": True} class RpcError(Exception): """A JSON-RPC protocol error, as opposed to a tool failure (which is a normal result with isError set, so the model can read it and act).""" def __init__(self, code: int, message: str): super().__init__(message) self.code = code def handle(g: Guaq, msg: dict): method = msg.get("method") if not isinstance(method, str): raise RpcError(-32600, "Invalid Request: no method") if method == "initialize": return {"protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": SERVER_INFO} if method == "tools/list": return {"tools": TOOLS} if method == "tools/call": params = msg.get("params") or {} name = params.get("name", "") args = params.get("arguments") or {} try: return _result(call_tool(g, name, args)) except GuaqError as e: hint = "" if e.payload.get("upgrade_required"): hint = (" This needs a higher plan — Ultimate for bulk export, an " "organisation seat for market signals. Contact hello@guaqai.me.") elif e.status == 429: hint = " The daily quota resets at 00:00 UTC." elif e.status == 401: hint = " Set GUAQ_API_KEY to a key from https://jobs.guaqai.me/profile." return _error(f"{e}{hint}") except Exception as e: return _error(f"{type(e).__name__}: {e}") if method == "ping": return {} if method.startswith("notifications/"): return None # notifications (notifications/initialized) get no reply raise RpcError(-32601, f"Method not found: {method}") def _reply(stdout, msg_id, result=None, error=None) -> None: out = {"jsonrpc": "2.0", "id": msg_id} if error is not None: out["error"] = error else: out["result"] = result stdout.write(json.dumps(out) + "\n") stdout.flush() def serve(stdin=None, stdout=None, client=None) -> None: stdin = stdin or sys.stdin stdout = stdout or sys.stdout g = client or Guaq(os.getenv("GUAQ_API_KEY", ""), os.getenv("GUAQ_API_BASE", "")) for line in stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except json.JSONDecodeError: _reply(stdout, None, error={"code": -32700, "message": "Parse error"}) continue if not isinstance(msg, dict): _reply(stdout, None, error={"code": -32600, "message": "Invalid Request"}) continue msg_id = msg.get("id") try: result, error = handle(g, msg), None except RpcError as e: result, error = None, {"code": e.code, "message": str(e)} except Exception: traceback.print_exc(file=sys.stderr) result, error = None, {"code": -32603, "message": "Internal error"} # A notification has no id and must get no response at all — replying to one is # a protocol violation that some hosts treat as a fatal error. if msg_id is None: continue _reply(stdout, msg_id, result, error) if __name__ == "__main__": serve()