#!/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 # Newest first. initialize echoes the client's version when we know it, else our newest, as # the MCP lifecycle spec requires; the stdio server and the remote /mcp endpoint share this. SUPPORTED_PROTOCOL_VERSIONS = ["2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05"] PROTOCOL_VERSION = SUPPORTED_PROTOCOL_VERSIONS[0] # icons/websiteUrl (MCP Implementation fields): without them a client falls back to a favicon # lookup, and Claude looks up the registrable domain guaqai.me, which is the Guaq AI company # mark rather than Employed's. SERVER_INFO = {"name": "guaq-jobs", "title": "Employed by Guaq AI", "version": "1.2.1", "websiteUrl": "https://jobs.guaqai.me", "icons": [{"src": "https://jobs.guaqai.me/assets/brand/logo-square-512.png", "mimeType": "image/png", "sizes": ["512x512"]}, {"src": "https://jobs.guaqai.me/assets/brand/logo-square.svg", "mimeType": "image/svg+xml", "sizes": ["any"]}]} # Server instructions and every tool description describe; they never direct. No "ask the # person first", "only when they ask", "call X before Y", no pricing or upgrade prompts, # nothing about other servers. OpenAI's review held the 2026-10-13 instructions for "further # review" over one such sentence, and Claude's directory makes the submitter attest that tool # descriptions carry no instructions about model behaviour. Confirmation before a write comes # from the annotations (readOnlyHint/destructiveHint), which the hosts act on themselves. INSTRUCTIONS = ( "Employed indexes live job postings read from about 5,000 employers' own " "applicant-tracking systems. " "Job search: count_jobs answers 'how many' questions, search_jobs lists matching roles, " "get_job opens one posting, list_companies and company_profile describe employers, and " "api_meta lists the accepted filter values. Visa sponsorship marked 'likely' is an " "inference from the posting, not a confirmation. Each posting's url is the employer's " "own application page. " "The signed-in person's account: get_my_profile, profile_options and update_my_profile " "hold their job-search profile; get_my_feed is their feed ranked against it; " "list_my_applications, track_job and untrack_job are their application tracker; " "get_my_alerts, follow_company, unfollow_company and set_email_alerts are their " "new-role alerts. update_my_profile accepts the ids that profile_options lists. " "These tools change only that person's own data. No tool applies to jobs, contacts " "employers, or changes billing.") # 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", "title": "Search live jobs", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to find specific open roles. Searches live job postings " "read from ~5,000 employers' own application systems and returns " "at most 50 rows (title, employer, location, seniority, posted " "date, visa signal, apply url) plus has_more/next_cursor. " "count_jobs gives the total for the same filters.", "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", "title": "Count matching jobs", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this when the question is 'how many' or 'what share' rather " "than about specific roles. Returns the number of live postings " "matching a set of filters, without listing them.", "inputSchema": {"type": "object", "properties": dict(_FILTER_SCHEMA)}, }, { "name": "get_job", "title": "Get one job posting", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to read one posting in full by its numeric id (the id " "field of a search_jobs row).", "inputSchema": {"type": "object", "properties": {"id": {"type": "integer"}}, "required": ["id"]}, }, { "name": "list_companies", "title": "List hiring employers", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to see which employers are hiring. Returns employers " "with their live role counts, most roles first. 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", "title": "Employer hiring profile", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to research one employer. Returns its hiring profile: " "open roles, seniority mix, function mix, top cities, " "technologies, longest-open roles, and the flow of postings over " "recent weeks.", "inputSchema": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}, }, { "name": "market_signal", "title": "Hiring market signals", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Aggregate hiring-market intelligence derived from the whole " "index, for analysts. Only available to organisation accounts; " "other accounts get an explanatory message. 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", "title": "Export jobs to a file", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False}, "description": "Local stdio server only (not offered by the hosted connector). " "Write every matching posting to a file on this machine (jsonl, " "csv or parquet) for analysis of the whole matching set. " "Requires an eligible plan.", "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": "get_my_profile", "title": "Your Employed profile", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to read the signed-in person's saved job-search profile: " "stage (student or working), education, graduation date, target " "categories, skills, cities, countries, work authorisation, " "employment types, experience range, plan and whether onboarding " "is complete.", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "profile_options", "title": "Profile field values", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to look up the exact ids each profile field accepts " "(the values update_my_profile takes): category paths with their names, country " "codes with the work-authorisation statuses offered there, " "employment types, education levels, cities and tracker statuses.", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "update_my_profile", "title": "Update your profile", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to set or change the signed-in person's profile, which " "decides what their feed shows. It overwrites the saved value of " "every field it is given and keeps every field it is not given. " "Values are the ids from profile_options; an unknown id is refused " "with the reason. A list field is replaced as a whole (the full list " "of cities, not just the new one). completed=true marks onboarding " "finished, so the web app skips it.", "inputSchema": { "type": "object", "additionalProperties": False, "properties": { "stage": {"type": "string", "enum": ["in_college", "professional"], "description": "Still studying, or already working."}, "college": {"type": "string"}, "branch": {"type": "string", "description": "Degree or major."}, "field_of_study": {"type": "string"}, "cgpa": {"type": "number"}, "education": {"type": "string", "enum": ["high_school", "diploma", "bachelors", "masters", "phd"], "description": "Highest qualification held or being studied for."}, "education_status": {"type": "string", "enum": ["pursuing", "completed"]}, "grad_month": {"type": "integer", "minimum": 1, "maximum": 12}, "grad_year": {"type": "integer", "minimum": 2000, "maximum": 2100}, "exp": {"type": "array", "items": {"type": "number"}, "minItems": 2, "maxItems": 2, "description": "[min, max] years of experience wanted, e.g. [0, 2]."}, "categories": {"type": "array", "items": {"type": "string"}, "description": "Category paths from profile_options, e.g. engineering/software/backend."}, "skills": {"type": "array", "items": {"type": "string"}}, "cities": {"type": "array", "items": {"type": "string"}, "description": "City names; synonyms match (bangalore finds Bengaluru)."}, "location_countries": {"type": "array", "items": {"type": "string"}, "description": "Country codes to work in: in, us, uk, de, sg, ..."}, "work_auth": {"type": "object", "additionalProperties": {"type": "array", "items": {"type": "string"}}, "description": "Country code -> statuses held there, e.g. " "{\"in\": [\"citizen\"], \"us\": [\"need_sponsorship\"]}."}, "routes": {"type": "array", "items": {"type": "string", "enum": ["authorized", "sponsored", "borderless"]}, "description": "How they could take a job: where already authorised, with visa sponsorship, or remote across borders."}, "employment_types": {"type": "array", "items": {"type": "string", "enum": EMPLOYMENT_TYPES}}, "modes": {"type": "array", "items": {"type": "string", "enum": ["remote", "office"]}, "description": "Work mode preference; office includes hybrid."}, "leadership_track": {"type": "string", "enum": ["ic", "manager"]}, "pay": {"type": "array", "items": {"type": "number"}, "minItems": 2, "maxItems": 2, "description": "[min, max] expected pay in thousands per year."}, "completed": {"type": "boolean", "description": "Mark onboarding finished."}, }, }, }, { "name": "get_my_feed", "title": "Your ranked job feed", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this when the person asks what roles suit them. Returns their " "personal feed: live roles they are eligible for, ranked against " "their saved profile, with a match score. Some accounts see only " "the top rows; the result says how many more there are.", "inputSchema": {"type": "object", "properties": { "q": {"type": "string", "description": "Optional text to match in title or company."}, "sort": {"type": "string", "enum": ["match", "fresh"], "description": "Best match (default) or newest."}, "posted_within_hours": {"type": "number", "description": "Only roles posted in the last N hours."}, "page": {"type": "integer", "description": "0-based page, for the next page of the feed."}, "limit": {"type": "integer", "description": "Rows to return, 1-25. Default 10."}, }}, }, { "name": "list_my_applications", "title": "Your saved and applied roles", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to read the person's tracker (the Saved tab): every role " "they saved or applied to, its current status, notes and history.", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "track_job", "title": "Save a role or set its status", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": False, "openWorldHint": False}, "description": "Use this to save a role to the person's tracker or record progress " "on it: saved, applied, oa, phone_screen, technical_interview, " "behavioral_interview, onsite, offer, rejected, withdrawn, ghosted, " "not_applied. On a role already tracked it replaces the current status " "(earlier statuses stay in the role's history).", "inputSchema": {"type": "object", "properties": { "job_id": {"type": "integer", "description": "The id of a search_jobs or get_my_feed row."}, "status": {"type": "string", "enum": ["saved", "applied", "oa", "phone_screen", "technical_interview", "behavioral_interview", "onsite", "offer", "rejected", "withdrawn", "ghosted", "not_applied"], "description": "Default saved."}, "remarks": {"type": "string", "description": "Optional note, e.g. the interviewer's name."}, }, "required": ["job_id"]}, }, { "name": "untrack_job", "title": "Remove a role from your tracker", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to remove a role, and its status history, from the " "person's tracker.", "inputSchema": {"type": "object", "properties": {"job_id": {"type": "integer"}}, "required": ["job_id"]}, }, { "name": "get_my_alerts", "title": "Your company and email alerts", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to see which employers the person follows for new-role " "alerts and whether email alerts are on.", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "follow_company", "title": "Follow an employer", "annotations": {"readOnlyHint": False, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to alert the person when an employer posts new roles. " "If the person's account does not include alerts, the reply says so.", "inputSchema": {"type": "object", "properties": {"company": {"type": "string"}}, "required": ["company"]}, }, { "name": "unfollow_company", "title": "Stop following an employer", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to stop new-role alerts for an employer.", "inputSchema": {"type": "object", "properties": {"company": {"type": "string"}}, "required": ["company"]}, }, { "name": "set_email_alerts", "title": "Turn email alerts on or off", "annotations": {"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": False}, "description": "Use this to turn the person's email digest of new matching roles " "on or off, replacing the current setting. If the person's account " "does not include email alerts, the reply says so.", "inputSchema": {"type": "object", "properties": {"enabled": {"type": "boolean"}}, "required": ["enabled"]}, }, { "name": "api_meta", "title": "Account, quota and filter values", "annotations": {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False}, "description": "Returns the caller's plan, remaining daily quota, and the exact " "vocabularies the filters accept (domains, role types, cities).", "inputSchema": {"type": "object", "properties": {}}, }, ] # The signed-in person's own account (/v1/me): free on every plan, never charged to the # quota, and the only tools that write. Kept as a set so hosts and tests can tell them apart. ACCOUNT_TOOLS = frozenset({"get_my_profile", "profile_options", "update_my_profile", "get_my_feed", "list_my_applications", "track_job", "untrack_job", "get_my_alerts", "follow_company", "unfollow_company", "set_email_alerts"}) # Output schemas. Deliberately loose (additionalProperties stays open): they tell a host # which keys to expect and render, without turning a harmless new field into a validation # failure for every connected client. Each tool result carries a matching structuredContent. _JOB_ROW_SCHEMA = {"type": "object", "properties": { "id": {"type": "integer"}, "company": {"type": "string"}, "title": {"type": "string"}, "location": {"type": "string"}, "location_scope": {"type": "string"}, "role_type": {"type": "string"}, "seniority_label": {"type": "string"}, "employment_type": {"type": "string"}, "yoe_min": {"type": "number"}, "work_arrangement": {"type": "string"}, "category_path": {"type": "string"}, "tech_stack": {"type": "array", "items": {"type": "string"}}, "posted_at": {"type": "string"}, "url": {"type": "string"}, "visa_supported": {"type": "boolean"}, "visa_likelihood": {"type": "number"}}} _OUTPUT_SCHEMAS = { "search_jobs": {"type": "object", "properties": { "showing": {"type": "integer"}, "has_more": {"type": "boolean"}, "next_cursor": {"type": ["integer", "string", "null"]}, "jobs": {"type": "array", "items": _JOB_ROW_SCHEMA}, "note": {"type": ["string", "null"]}}, "required": ["showing", "jobs"]}, "count_jobs": {"type": "object", "properties": {"count": {"type": "integer"}}, "required": ["count"]}, "get_job": _JOB_ROW_SCHEMA, "list_companies": {"type": "object", "properties": {"companies": {"type": "array", "items": { "type": "object", "properties": {"company": {"type": "string"}, "open_roles": {"type": "integer"}}}}}, "required": ["companies"]}, "company_profile": {"type": "object", "properties": { "company": {"type": "string"}, "open_roles": {"type": "integer"}, "seniority_mix": {"type": "object"}, "function_mix": {"type": "object"}, "top_cities": {"type": "object"}, "tech_stack": {"type": "object"}, "longest_open_roles": {"type": "array"}, "posting_flow": {"type": "object"}}, "required": ["company"]}, "market_signal": {"type": "object"}, "api_meta": {"type": "object"}, "export_jobs": {"type": "object", "properties": {"written": {"type": "string"}, "bytes": {"type": "integer"}}}, } _OUTPUT_SCHEMAS.update({ "get_my_profile": {"type": "object", "properties": { "plan_tier": {"type": "string"}, "onboarding_completed": {"type": "boolean"}, "preferences": {"type": "object"}}}, "profile_options": {"type": "object"}, "update_my_profile": {"type": "object", "properties": { "saved": {"type": "boolean"}, "preferences": {"type": "object"}}, "required": ["saved"]}, "get_my_feed": {"type": "object", "properties": { "showing": {"type": "integer"}, "total": {"type": "integer"}, "has_more": {"type": "boolean"}, "next_page": {"type": ["integer", "null"]}, "jobs": {"type": "array", "items": _JOB_ROW_SCHEMA}, "note": {"type": ["string", "null"]}}, "required": ["showing", "jobs"]}, "list_my_applications": {"type": "object", "properties": { "applications": {"type": "array", "items": {"type": "object"}}}, "required": ["applications"]}, "track_job": {"type": "object", "properties": { "job_id": {"type": "integer"}, "status": {"type": "string"}, "history": {"type": "array"}}}, "untrack_job": {"type": "object", "properties": {"removed": {"type": "boolean"}}}, "get_my_alerts": {"type": "object", "properties": { "companies": {"type": "array"}, "email_alerts": {"type": "object"}}}, "follow_company": {"type": "object", "properties": {"companies": {"type": "array"}}}, "unfollow_company": {"type": "object", "properties": {"companies": {"type": "array"}}}, "set_email_alerts": {"type": "object", "properties": {"enabled": {"type": "boolean"}}}, }) def _allow_null(schema: dict) -> dict: """Every declared field may also be null. A posting's location, seniority, posted date or visa signal is often unknown, and get_job returns the full row with those as null; a host that validates structuredContent against outputSchema (the MCP SDKs do) turned each such result into a tool error (MCPJam swarm 2026-10-12: get_job failed 72 times). The result itself stays an object; only its fields, and the fields of objects inside it, widen.""" out = dict(schema) if isinstance(out.get("properties"), dict): props = {} for key, sub in out["properties"].items(): sub = _allow_null(sub) kind = sub.get("type") if isinstance(kind, str) and kind != "null": sub["type"] = [kind, "null"] elif isinstance(kind, list) and "null" not in kind: sub["type"] = kind + ["null"] props[key] = sub out["properties"] = props if isinstance(out.get("items"), dict): out["items"] = _allow_null(out["items"]) return out _OUTPUT_SCHEMAS = {name: _allow_null(s) for name, s in _OUTPUT_SCHEMAS.items()} for _t in TOOLS: _t["outputSchema"] = _OUTPUT_SCHEMAS[_t["name"]] _t["annotations"]["title"] = _t["title"] # 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", "visa_supported", "visa_likelihood", "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 _feed_row(card: dict) -> dict: """A feed card (the web app's shape) as a search_jobs row, so the same view renders it.""" mins = card.get("mins") posted = None if card.get("posted_known") and isinstance(mins, (int, float)): import datetime as _dt posted = (_dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(minutes=mins)).date().isoformat() mode = (card.get("mode") or "").lower().replace("onsite", "on_site") row = {"id": card.get("id"), "company": card.get("co"), "title": card.get("role"), "location": card.get("city"), "seniority_label": card.get("level"), "employment_type": card.get("employment"), "yoe_min": card.get("yoe_min"), "work_arrangement": mode or None, "category_path": card.get("category") or None, "tech_stack": card.get("tech_stack") or None, "posted_at": posted, "visa_supported": card.get("visa"), "visa_likelihood": card.get("visa_odds"), "url": card.get("url"), "match": card.get("match"), "salary": card.get("salary")} return {k: v for k, v in row.items() if v not in (None, "", [])} def _profile_view(p: dict) -> dict: """The profile without the web app's bookkeeping keys.""" prefs = dict(p.get("preferences") or {}) for k in ("plan_tier", "is_pro", "referral_code", "coins", "friends_joined", "onboarding_completed", "platform_role", "organization", "telegram"): prefs.pop(k, None) return {"plan_tier": p.get("plan_tier"), "onboarding_completed": p.get("onboarding_completed"), "preferences": prefs, "edit_in_browser": "https://jobs.guaqai.me/profile"} def call_tool(g: Guaq, name: str, args: dict): if name == "get_my_profile": return _profile_view(g.my_profile()) if name == "profile_options": return g.profile_options() if name == "update_my_profile": fields = {k: v for k, v in args.items() if v is not None} if not fields: raise GuaqError("Nothing to update: pass at least one profile field.", 400) # "changed" is what actually differs afterwards, read back from the store: an agent # resending a value the profile already holds (the live test of 2026-10-13 sent # cities=[] to an empty list) must not be told it changed something. before = _profile_view(g.my_profile()) g.update_my_profile(**fields) out = _profile_view(g.my_profile()) def stored(view, k): return view.get("onboarding_completed") if k == "completed" else view["preferences"].get(k) changed = sorted(k for k in fields if stored(before, k) != stored(out, k)) return {"saved": True, "changed": changed, "unchanged": sorted(k for k in fields if k not in changed), **out} if name == "get_my_feed": limit = max(1, min(int(args.get("limit") or 10), 25)) page = g.my_feed(search=args.get("q"), sort_by=args.get("sort") or "match", posted_within=args.get("posted_within_hours"), page=args.get("page"), page_size=max(10, limit)) rows = [_feed_row(c) for c in (page.get("jobs") or [])][:limit] more = bool(page.get("has_more")) or len(page.get("jobs") or []) > limit note = None if page.get("paywall_triggered"): note = (f"This account's feed shows the top {page.get('free_tier_rows')} of " f"{page.get('total_count')} matching roles.") elif page.get("widened"): note = "Few exact matches, so the feed widened to similar roles." return {"showing": len(rows), "total": page.get("total_count"), "has_more": more, "next_page": (int(args.get("page") or 0) + 1) if page.get("has_more") else None, "jobs": rows, "note": note} if name == "list_my_applications": apps = g.my_applications().get("applications") or [] return {"applications": [{k: a.get(k) for k in ("job_id", "title", "company", "location", "url", "status", "status_label", "updated_at", "history")} for a in apps]} if name == "track_job": res = g.track(args["job_id"], status=args.get("status") or "saved", remarks=args.get("remarks") or "") return {"job_id": int(args["job_id"]), "status": res.get("status"), "history": res.get("history")} if name == "untrack_job": return {"job_id": int(args["job_id"]), "removed": bool(g.untrack(args["job_id"]).get("removed"))} if name == "get_my_alerts": ca, ea = g.company_alerts(), g.email_alerts() return {"companies": ca.get("tracked_details") or ca.get("tracked") or [], "company_alerts_available": ca.get("entitled"), "max_companies": ca.get("max_tracked"), "email_alerts": ea} if name in ("follow_company", "unfollow_company"): fn = g.follow_company if name == "follow_company" else g.unfollow_company res = fn(args["company"]) return {"companies": res.get("tracked_details") or res.get("tracked") or []} if name == "set_email_alerts": return g.set_email_alerts(bool(args.get("enabled"))) 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"), "filters": _filters(args), "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) # A tool result larger than this is cut for the text block (the model's context); the # structured copy is capped separately by the tools themselves (50 rows, trimmed fields). MAX_TEXT_CHARS = 60_000 def _result(payload) -> dict: text = json.dumps(payload, indent=2, default=str) if len(text) > MAX_TEXT_CHARS: text = text[:MAX_TEXT_CHARS] + "\n... [truncated; narrow the filters for the rest]" out = {"content": [{"type": "text", "text": text}]} if isinstance(payload, dict): # json round trip: structuredContent must be plain JSON (dates, Decimals -> text). out["structuredContent"] = json.loads(json.dumps(payload, default=str)) return out 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": wanted = (msg.get("params") or {}).get("protocolVersion") return {"protocolVersion": wanted if wanted in SUPPORTED_PROTOCOL_VERSIONS else PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": SERVER_INFO, "instructions": INSTRUCTIONS} 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 {} spec = next((t for t in TOOLS if t["name"] == name), None) if spec is None: return _error(f"Unknown tool: {name}") missing = [k for k in spec["inputSchema"].get("required", []) if args.get(k) in (None, "")] if missing: return _error("Missing required argument: " + ", ".join(missing)) try: return _result(call_tool(g, name, args)) except GuaqError as e: hint = "" # Factual, not promotional: ChatGPT's review rejects upgrade prompts in a tool # reply. The account page is where the person manages their own plan. if e.payload.get("trial_exhausted") or e.payload.get("tier") == "free": hint = " Account details: https://jobs.guaqai.me/profile" elif e.payload.get("upgrade_required") and name in ACCOUNT_TOOLS: hint = " Account details: https://jobs.guaqai.me/profile" elif 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 = " Run 'npx -y employed-mcp login' (or click Authenticate in your agent) or set GUAQ_API_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__": if len(sys.argv) > 1 and sys.argv[1] in ("-h", "--help", "help"): print("Employed MCP Server (JSON-RPC 2.0 over stdio)") print("Usage: guaq_mcp.py (reads JSON-RPC messages from stdin, writes to stdout)") print("Environment variables: GUAQ_API_KEY, GUAQ_API_BASE") sys.exit(0) if len(sys.argv) > 1 and sys.argv[1] in ("login", "auth"): from guaq_client import run_oauth_flow base = os.getenv("GUAQ_API_BASE", "https://jobs.guaqai.me") sys.exit(0 if run_oauth_flow(base) else 1) serve()