spencerm/freshservice-claude
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ff627b5620c1e9ade0743417d8a1dfcee4bd6e57
freshservice-claude/mcp-server/server.py
spencerm e38841918d Fix status-cache poisoning, add HTTP timeouts, coerce custom status names 
- _status_choices no longer caches failures: a transient /ticket_form_fields
  error used to cache [] forever, silently degrading unresolved=True to an
  unfiltered listing (including Resolved/Closed) until container restart.
  unresolved=True and list_ticket_statuses now return an explicit error when
  the status list can't be fetched instead of degrading silently.
- All API calls (server and skill helper) now use a 30s timeout with a clean
  error, so a hung connection can't block a tool call indefinitely.
- _coerce_status falls back to the instance's live status list, so custom
  names like "Working" coerce to their id instead of 400ing at the API.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-08 10:38:56 -06:00

471 lines
19 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.
Env:
FRESHSERVICE_DOMAIN "acme" or "acme.freshservice.com"
FRESHSERVICE_API_KEY profile API key
MCP_AUTH_TOKEN shared bearer token the client must send (optional but
strongly recommended; if unset, auth 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.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 _config() -> tuple[str, str]:
domain = os.environ.get("FRESHSERVICE_DOMAIN", "").strip()
api_key = os.environ.get("FRESHSERVICE_API_KEY", "").strip()
if not domain or not api_key:
raise RuntimeError(
"Missing credentials. Set FRESHSERVICE_DOMAIN and FRESHSERVICE_API_KEY."
)
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