Add Freshservice MCP server for Cowork deployment
Cowork runs the agent in an isolated VM, so the uploaded skill can't see host credentials or reach the Freshservice host. Add an HTTP/streamable MCP server (bearer-auth, Dockerfile, compose) to run as a container on the LAN box and connect via mcp-remote like the existing sharepoint-lists server. The skill remains the right tool for the non-sandboxed Claude Code CLI. Also gitignore mcp-server/.env and include the desktop skill install package. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
365
mcp-server/server.py
Normal file
365
mcp-server/server.py
Normal file
@@ -0,0 +1,365 @@
|
||||
#!/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
|
||||
|
||||
mcp = FastMCP("freshservice")
|
||||
|
||||
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
|
||||
|
||||
|
||||
# --- Tickets ---------------------------------------------------------------
|
||||
@mcp.tool()
|
||||
def list_tickets(query: str | None = None, updated_since: str | None = None,
|
||||
order_by: str | None = None, order_type: str | None = None,
|
||||
page: int = 1, per_page: int = 30) -> str:
|
||||
"""List or filter tickets. With `query`, uses /tickets/filter, e.g.
|
||||
"status:2 AND priority:3". Status 2=Open 3=Pending 4=Resolved 5=Closed;
|
||||
priority 1=Low 2=Medium 3=High 4=Urgent."""
|
||||
if query:
|
||||
params = {"query": f'"{query}"', "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, "page": page, "per_page": min(per_page, 100)}
|
||||
return _out(_request("GET", "/tickets", params=params))
|
||||
|
||||
|
||||
@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
Block a user