spencerm/freshservice-claude
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4f9a862c25ccff4938a0dc1e25b184be17f53b40
freshservice-claude/mcp-server/server.py
spencerm 4f9a862c25 Support per-user identity via request headers 
Add multi-user auth: _config reads X-Freshservice-Domain/X-Freshservice-Key
from the per-request headers (via the SDK request_ctx, set inside the tool's
task) and prefers them over the container env. One shared server can now act as
each caller's own Freshservice identity — actions attributed to them, their
permissions, their rate limit — while the env vars remain a single-user
fallback. Verified end-to-end that concurrent clients don't cross-wire and that
a client with no creds errors instead of falling back silently.

Docs/.env.example updated with the multi-user client config and the guidance to
omit env FS creds on a shared deployment.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-10 08:23:52 -06:00

509 lines
21 KiB
Python
Raw Blame History

#!/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)]
# --- 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)}
return _out(_request("GET", "/tickets/filter", params=params))
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)}
return _out(_request("GET", "/tickets", params=params))
@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` e.g. "conversations,requester,stats"."""
params = {"include": include} if include else None
return _out(_request("GET", f"/tickets/{ticket_id}", params=params))
@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()
Reference in New Issue View Git Blame Copy Permalink