spencerm
0b6747812f
The pesco instance defines 13 ticket statuses, not the 4 defaults, so a (status:2 OR status:3) 'open tickets' filter silently dropped Working, Scheduled, Waiting*, Pending Approval, etc. — undercounting 24 open tickets as 1. Add a status helper that reads the status field from /ticket_form_fields and treats only ids 4/5 (and closed/resolved-labelled) as terminal. Expose list_ticket_statuses() and a list_tickets(unresolved=True) shortcut that builds the full non-terminal status clause automatically. Document the gotcha in the skill reference. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
446 lines
18 KiB
Python
446 lines
18 KiB
Python
#!/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) 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}
|
|
|
|
|
|
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
|
|
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."""
|
|
if _status_cache["choices"] is None:
|
|
res = _request("GET", "/ticket_form_fields")
|
|
choices: list[dict[str, Any]] = []
|
|
if res.get("ok"):
|
|
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
|
|
_status_cache["choices"] = choices
|
|
return _status_cache["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 ids:
|
|
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)."""
|
|
statuses = [{"id": c.get("id"), "label": c.get("value"),
|
|
"terminal": _is_terminal(c)} for c in _status_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()
|