#!/usr/bin/env python3 """Freshservice MCP server (streamable-HTTP, bearer-auth). Runs as a container on the LAN box and is reached from Claude Desktop / Cowork via mcp-remote, exactly like the sibling sharepoint-lists server. Credentials live here (in the container), outside the Cowork VM sandbox. Identity / credentials, in precedence order: 1. Per-request headers (multi-user): each client sends its OWN X-Freshservice-Domain and X-Freshservice-Key on every call, so the shared server acts as that caller's identity and actions are attributed to them. Use this when more than one person shares the server. 2. Env vars (single-user): FRESHSERVICE_DOMAIN / FRESHSERVICE_API_KEY. A fallback used only when the headers are absent. OMIT these on a multi-user deployment so a client that forgets its headers errors out rather than silently acting as whoever owns the env key. Env: FRESHSERVICE_DOMAIN "acme" or "acme.freshservice.com" (single-user only) FRESHSERVICE_API_KEY profile API key (single-user only) MCP_AUTH_TOKEN shared bearer token every client must send to reach the server (gate); separate from per-user identity. Strongly recommended; if unset, the server is open. MCP_TRANSPORT "http" (default here) or "stdio" HOST / PORT bind address (default 0.0.0.0:3838) Docs: https://api.freshservice.com """ from __future__ import annotations import base64 import json import os import re import urllib.error import urllib.parse import urllib.request from typing import Any from mcp.server.fastmcp import FastMCP from mcp.server.lowlevel.server import request_ctx from mcp.server.transport_security import TransportSecuritySettings # The streamable-HTTP transport enables DNS-rebinding protection by default, # which validates the Host header against an allow-list (localhost only when # empty) and returns HTTP 421 "Invalid Host header" otherwise. This server is # reached by LAN IP (e.g. 192.168.101.12:3839) from mcp-remote — not a browser — # and is already gated by a bearer token, so the protection only breaks the # connection. Disable it. mcp = FastMCP( "freshservice", transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False), ) TICKET_STATUS = {2: "Open", 3: "Pending", 4: "Resolved", 5: "Closed"} TICKET_PRIORITY = {1: "Low", 2: "Medium", 3: "High", 4: "Urgent"} def _header_creds() -> tuple[str, str]: """Per-request Freshservice creds from headers the client sends on every call — X-Freshservice-Domain / X-Freshservice-Key — so one shared server acts as each caller's own identity. Returns ("", "") when unavailable (stdio transport, or headers absent). Read via the SDK's request_ctx, which is set inside the tool's own task, so it reflects THIS call's request.""" try: rc = request_ctx.get() except LookupError: return "", "" req = getattr(rc, "request", None) if req is None: return "", "" h = req.headers return (h.get("x-freshservice-domain", "").strip(), h.get("x-freshservice-key", "").strip()) def _config() -> tuple[str, str]: # Per-request header creds take precedence over the container env, so a # shared server acts as each caller's identity. Env is the single-user # fallback — OMIT FRESHSERVICE_* from a multi-user deployment so a client # that forgets its headers errors out instead of silently acting as whoever # owns the env key. h_domain, h_key = _header_creds() domain = (h_domain or os.environ.get("FRESHSERVICE_DOMAIN", "")).strip() api_key = (h_key or os.environ.get("FRESHSERVICE_API_KEY", "")).strip() if not domain or not api_key: raise RuntimeError( "Missing credentials. Send X-Freshservice-Domain and " "X-Freshservice-Key headers (per-user), or set FRESHSERVICE_DOMAIN " "and FRESHSERVICE_API_KEY in the environment (single-user)." ) if "." not in domain: domain = f"{domain}.freshservice.com" return domain, api_key def _request(method: str, path: str, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None) -> dict[str, Any]: domain, api_key = _config() path = path if path.startswith("/") else f"/{path}" if not path.startswith("/api/"): path = f"/api/v2{path}" url = f"https://{domain}{path}" if params: clean = {k: v for k, v in params.items() if v is not None} if clean: url += "?" + urllib.parse.urlencode(clean) token = base64.b64encode(f"{api_key}:X".encode()).decode() headers = {"Authorization": f"Basic {token}", "Content-Type": "application/json", "Accept": "application/json"} data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method.upper(), headers=headers) try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read().decode() result: dict[str, Any] = {"ok": True, "status": resp.status, "data": json.loads(raw) if raw else {}} link = resp.headers.get("Link") if link: m = re.search(r'page=(\d+)[^>]*>;\s*rel="next"', link) if m: result["next_page"] = int(m.group(1)) rem = resp.headers.get("X-RateLimit-Remaining") if rem is not None: result["rate_limit_remaining"] = rem return result except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace") try: detail = json.loads(detail) except json.JSONDecodeError: pass return {"ok": False, "status": e.code, "error": detail, "url": url, "method": method.upper()} except urllib.error.URLError as e: return {"ok": False, "error": f"Network error: {e.reason}", "url": url} except TimeoutError: return {"ok": False, "error": "Request timed out after 30s", "url": url} def _out(result: dict[str, Any]) -> str: return json.dumps(result, indent=2, default=str) def _coerce_status(v: Any) -> Any: if isinstance(v, str): for k, name in TICKET_STATUS.items(): if name.lower() == v.lower(): return k # Not a default name — try the instance's full status list, which # includes custom statuses like "Working" or "Waiting Vendor Support". for c in _status_choices(): if str(c.get("value", "")).lower() == v.lower(): return c["id"] return v def _coerce_priority(v: Any) -> Any: if isinstance(v, str): for k, name in TICKET_PRIORITY.items(): if name.lower() == v.lower(): return k return v # Freshservice instances can define many custom statuses beyond the 4 defaults. # Resolved and Closed are always system ids 4 and 5; everything else (including # custom statuses like "Working" or "Waiting Vendor Support") is "open". _status_cache: dict[str, Any] = {"choices": None} def _status_choices() -> list[dict[str, Any]]: """Fetch (and cache) the instance's ticket status choices via the status field on /ticket_form_fields. Only a successful, non-empty result is cached: a transient API failure must not permanently disable status-aware features (unresolved=True would otherwise silently degrade to an unfiltered listing until restart).""" if _status_cache["choices"] is not None: return _status_cache["choices"] res = _request("GET", "/ticket_form_fields") if not res.get("ok"): return [] choices: list[dict[str, Any]] = [] for f in res["data"].get("ticket_fields", []): if f.get("name") == "status" or f.get("field_type") == "default_status": choices = f.get("choices", []) break if choices: _status_cache["choices"] = choices return choices def _is_terminal(choice: dict[str, Any]) -> bool: """A status counts as terminal (resolved/closed) if it's system id 4/5 or its label says so (covers custom 'Closed - …' statuses).""" label = str(choice.get("value", "")).lower() return choice.get("id") in (4, 5) or "resolved" in label or "closed" in label def _open_status_ids() -> list[int]: return [c["id"] for c in _status_choices() if c.get("id") is not None and not _is_terminal(c)] # Impact/urgency are fixed 1-3 scales in Freshservice (priority is fixed 1-4). IMPACT_URGENCY = {1: "Low", 2: "Medium", 3: "High"} def _label_ticket(t: Any) -> Any: """Add *_label fields next to a ticket's numeric enums (priority, status, impact, urgency) so no consumer ever guesses the scale. A weak model once read priority 4 and called it "Low (4)" — Urgent — in a customer-visible note; labels in the data remove that entire error class.""" if not isinstance(t, dict): return t p = t.get("priority") if p in TICKET_PRIORITY: t["priority_label"] = TICKET_PRIORITY[p] for field in ("impact", "urgency"): v = t.get(field) if v in IMPACT_URGENCY: t[f"{field}_label"] = IMPACT_URGENCY[v] s = t.get("status") if isinstance(s, int) and "status_name" not in t: label = next((c.get("value") for c in _status_choices() if c.get("id") == s), None) or TICKET_STATUS.get(s) if label: t["status_label"] = label return t # --- Tickets --------------------------------------------------------------- @mcp.tool() def list_tickets(query: str | None = None, unresolved: bool = False, updated_since: str | None = None, order_by: str | None = None, order_type: str | None = None, workspace_id: int | None = None, page: int = 1, per_page: int = 30) -> str: """List or filter tickets. With `query` (or `unresolved`), uses /tickets/filter. Default status codes: 2=Open 3=Pending 4=Resolved 5=Closed — but this instance may define CUSTOM statuses too (e.g. "Working", "Waiting Vendor Support"). Call list_ticket_statuses() to see them all. Priority: 1=Low 2=Medium 3=High 4=Urgent. `unresolved=True`: the turnkey "still-open" filter. The server fetches every status, drops Resolved/Closed (and any custom closed-like status), and builds the `(status:.. OR ..)` clause for you — so you don't miss custom statuses. Combine it with a `query` to scope further (they are AND-ed), e.g. list_tickets(query="agent_id:21000816864", unresolved=True) = my open tickets. Building a `query` by hand: * Filter by ASSIGNEE with the numeric `agent_id`, NOT an email — resolve via list_agents(email=...) first and read `id`. * There is no negation operator; prefer `unresolved=True` over hand-listing statuses. Quote strings: `tag:'vip'`. Dates: `created_at:>'2026-01-01'`. `workspace_id`: accounts with multiple workspaces only search one workspace by default. Pass a workspace id to target another (see an agent's workspace_ids). """ if unresolved: ids = _open_status_ids() if not ids: # Never degrade silently to an unfiltered listing (which would # include Resolved/Closed tickets) — surface the failure instead. return _out({"ok": False, "error": "unresolved=True needs this instance's status list, but " "fetching /ticket_form_fields failed. Retry, or build an " "explicit (status:.. OR ..) query using " "list_ticket_statuses()."}) clause = "(" + " OR ".join(f"status:{i}" for i in ids) + ")" query = f"({query}) AND {clause}" if query else clause if query: params = {"query": f'"{query}"', "workspace_id": workspace_id, "page": page, "per_page": min(per_page, 100)} res = _request("GET", "/tickets/filter", params=params) else: params = {"updated_since": updated_since, "order_by": order_by, "order_type": order_type, "workspace_id": workspace_id, "page": page, "per_page": min(per_page, 100)} res = _request("GET", "/tickets", params=params) if res.get("ok"): for t in res["data"].get("tickets", []): _label_ticket(t) return _out(res) @mcp.tool() def list_ticket_statuses() -> str: """List every ticket status defined in this Freshservice instance (defaults + custom), each flagged `terminal` (Resolved/Closed) or open. Use this to build status filters correctly, or just call list_tickets(unresolved=True).""" choices = _status_choices() if not choices: return _out({"ok": False, "error": "Could not fetch ticket statuses from /ticket_form_fields."}) statuses = [{"id": c.get("id"), "label": c.get("value"), "terminal": _is_terminal(c)} for c in choices] return _out({"ok": True, "statuses": statuses, "open_ids": _open_status_ids()}) @mcp.tool() def get_ticket(ticket_id: int, include: str | None = None) -> str: """Get a ticket by id. `include` accepts a comma-separated subset of: conversations, requester, requested_for, stats, problem, assets, tags, related_tickets. NOTE: include=assets is the ONLY way to see the ticket's associated CMDB assets (a /tickets/{id}/associated-assets path does not exist — it 404s); the assets come back embedded in the ticket object. Numeric enums come back with companion *_label fields (priority_label, status_label, impact_label, urgency_label) — use those names, never guess the scale (priority 4 is Urgent, not low).""" params = {"include": include} if include else None res = _request("GET", f"/tickets/{ticket_id}", params=params) if res.get("ok"): _label_ticket(res["data"].get("ticket")) return _out(res) @mcp.tool() def create_ticket(subject: str, description: str, email: str | None = None, requester_id: int | None = None, priority: Any = 1, status: Any = 2, group_id: int | None = None, agent_id: int | None = None, ticket_type: str | None = None, tags: list[str] | None = None, custom_fields: dict[str, Any] | None = None) -> str: """Create a ticket. Provide either `email` or `requester_id`.""" body: dict[str, Any] = {"subject": subject, "description": description, "priority": _coerce_priority(priority), "status": _coerce_status(status)} if requester_id is not None: body["requester_id"] = requester_id elif email: body["email"] = email if group_id is not None: body["group_id"] = group_id if agent_id is not None: body["responder_id"] = agent_id if ticket_type: body["type"] = ticket_type if tags: body["tags"] = tags if custom_fields: body["custom_fields"] = custom_fields return _out(_request("POST", "/tickets", body=body)) @mcp.tool() def update_ticket(ticket_id: int, fields: dict[str, Any]) -> str: """Update a ticket. Pass only fields to change (status/priority accept names).""" f = dict(fields) if "status" in f: f["status"] = _coerce_status(f["status"]) if "priority" in f: f["priority"] = _coerce_priority(f["priority"]) return _out(_request("PUT", f"/tickets/{ticket_id}", body=f)) @mcp.tool() def reply_ticket(ticket_id: int, body: str) -> str: """Post a public reply (emailed to the requester) on a ticket.""" return _out(_request("POST", f"/tickets/{ticket_id}/reply", body={"body": body})) @mcp.tool() def add_note(ticket_id: int, body: str, private: bool = True) -> str: """Add a note to a ticket. private=True is internal-only.""" return _out(_request("POST", f"/tickets/{ticket_id}/notes", body={"body": body, "private": private})) @mcp.tool() def list_ticket_conversations(ticket_id: int, page: int = 1, per_page: int = 30) -> str: """List the conversation thread (replies and notes) on a ticket.""" return _out(_request("GET", f"/tickets/{ticket_id}/conversations", params={"page": page, "per_page": min(per_page, 100)})) # --- Assets / CMDB --------------------------------------------------------- @mcp.tool() def search_assets(query: str | None = None, page: int = 1, per_page: int = 30) -> str: """List or search assets. `query` e.g. "name:'MacBook'" or "asset_type_id:7".""" if query: params = {"filter": f'"{query}"', "page": page, "per_page": min(per_page, 100)} else: params = {"page": page, "per_page": min(per_page, 100)} return _out(_request("GET", "/assets", params=params)) @mcp.tool() def get_asset(display_id: int, include: str | None = None) -> str: """Get an asset by display id. `include` e.g. "type_fields".""" params = {"include": include} if include else None return _out(_request("GET", f"/assets/{display_id}", params=params)) @mcp.tool() def update_asset(display_id: int, fields: dict[str, Any]) -> str: """Update an asset. Pass only the fields to change.""" return _out(_request("PUT", f"/assets/{display_id}", body=fields)) # --- People ---------------------------------------------------------------- @mcp.tool() def find_requester(email: str | None = None, query: str | None = None, page: int = 1, per_page: int = 30) -> str: """Find requesters by email or filter query (e.g. "department_id:123").""" if email: params = {"email": email, "page": page, "per_page": min(per_page, 100)} elif query: params = {"query": f'"{query}"', "page": page, "per_page": min(per_page, 100)} else: params = {"page": page, "per_page": min(per_page, 100)} return _out(_request("GET", "/requesters", params=params)) @mcp.tool() def list_agents(email: str | None = None, query: str | None = None, page: int = 1, per_page: int = 30) -> str: """List agents, optionally filtered by email or query (e.g. "group_id:123").""" if email: params = {"email": email, "page": page, "per_page": min(per_page, 100)} elif query: params = {"query": f'"{query}"', "page": page, "per_page": min(per_page, 100)} else: params = {"page": page, "per_page": min(per_page, 100)} return _out(_request("GET", "/agents", params=params)) @mcp.tool() def list_groups(page: int = 1, per_page: int = 30) -> str: """List agent groups.""" return _out(_request("GET", "/groups", params={"page": page, "per_page": min(per_page, 100)})) @mcp.tool() def list_departments(page: int = 1, per_page: int = 30) -> str: """List departments.""" return _out(_request("GET", "/departments", params={"page": page, "per_page": min(per_page, 100)})) # --- Change / Problem / Release -------------------------------------------- @mcp.tool() def list_changes(query: str | None = None, page: int = 1, per_page: int = 30) -> str: """List or filter changes. `query` e.g. "status:1 AND priority:3".""" if query: params = {"query": f'"{query}"', "page": page, "per_page": min(per_page, 100)} return _out(_request("GET", "/changes/filter", params=params)) return _out(_request("GET", "/changes", params={"page": page, "per_page": min(per_page, 100)})) @mcp.tool() def get_change(change_id: int, include: str | None = None) -> str: """Get a change record by id.""" params = {"include": include} if include else None return _out(_request("GET", f"/changes/{change_id}", params=params)) @mcp.tool() def create_change(subject: str, description: str, requester_id: int | None = None, email: str | None = None, priority: Any = 1, status: int = 1, planned_start_date: str | None = None, planned_end_date: str | None = None, custom_fields: dict[str, Any] | None = None) -> str: """Create a change record.""" body: dict[str, Any] = {"subject": subject, "description": description, "priority": _coerce_priority(priority), "status": status} if requester_id is not None: body["requester_id"] = requester_id elif email: body["email"] = email if planned_start_date: body["planned_start_date"] = planned_start_date if planned_end_date: body["planned_end_date"] = planned_end_date if custom_fields: body["custom_fields"] = custom_fields return _out(_request("POST", "/changes", body=body)) @mcp.tool() def list_problems(page: int = 1, per_page: int = 30) -> str: """List problem records.""" return _out(_request("GET", "/problems", params={"page": page, "per_page": min(per_page, 100)})) @mcp.tool() def get_problem(problem_id: int) -> str: """Get a problem record by id.""" return _out(_request("GET", f"/problems/{problem_id}")) @mcp.tool() def list_releases(page: int = 1, per_page: int = 30) -> str: """List release records.""" return _out(_request("GET", "/releases", params={"page": page, "per_page": min(per_page, 100)})) @mcp.tool() def get_release(release_id: int) -> str: """Get a release record by id.""" return _out(_request("GET", f"/releases/{release_id}")) # --- Generic escape hatch -------------------------------------------------- @mcp.tool() def freshservice_request(method: str, path: str, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None) -> str: """Call any Freshservice API v2 endpoint directly (solutions/KB, products, vendors, ticket_fields, time_entries, etc.). "/api/v2" is prepended if absent.""" return _out(_request(method, path, params=params, body=body)) def _build_http_app(): """Wrap the streamable-HTTP app with a static bearer-token guard.""" from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse expected = os.environ.get("MCP_AUTH_TOKEN", "").strip() class BearerAuth(BaseHTTPMiddleware): async def dispatch(self, request, call_next): if expected: header = request.headers.get("authorization", "") token = header[7:].strip() if header.lower().startswith("bearer ") else "" if token != expected: return JSONResponse({"error": "unauthorized"}, status_code=401) return await call_next(request) app = mcp.streamable_http_app() app.add_middleware(BearerAuth) return app if __name__ == "__main__": transport = os.environ.get("MCP_TRANSPORT", "http").lower() if transport in ("http", "streamable-http", "streamable_http"): import uvicorn mcp.settings.host = os.environ.get("HOST", "0.0.0.0") mcp.settings.port = int(os.environ.get("PORT", "3838")) uvicorn.run(_build_http_app(), host=mcp.settings.host, port=mcp.settings.port) else: mcp.run()