feat(zendesk): new Zendesk ticketing integration
Zendesk Support API v2, 7 commands: create/get/update/search tickets, add comment (public/internal), list users. API-token (Basic) auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,96 @@
|
||||
id: zendesk
|
||||
name: Zendesk
|
||||
version: 1.0.0
|
||||
description: "Zendesk Support (REST API v2) — ticketing: create, read, update and search tickets, add comments (public or internal), and list users. API-token (Basic) authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: create/get/update/search tickets, add comment, list users."
|
||||
category: ticketing
|
||||
|
||||
# Per-instance configuration. Auth is HTTP Basic with '<email>/token:<api_token>'.
|
||||
config_schema:
|
||||
properties:
|
||||
subdomain:
|
||||
type: string
|
||||
description: "Zendesk subdomain (the X in https://X.zendesk.com)"
|
||||
email:
|
||||
type: string
|
||||
description: "Agent email address"
|
||||
api_token:
|
||||
type: string
|
||||
description: "Zendesk API token"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- subdomain
|
||||
- email
|
||||
- api_token
|
||||
|
||||
commands:
|
||||
- id: create_ticket
|
||||
name: zendesk-create-ticket
|
||||
description: "Create a ticket."
|
||||
inputs_schema:
|
||||
properties:
|
||||
subject: { type: string, description: "Ticket subject" }
|
||||
comment: { type: string, description: "Initial comment / description" }
|
||||
priority: { type: string, description: "low, normal, high, or urgent" }
|
||||
type: { type: string, description: "problem, incident, question, or task" }
|
||||
tags: { type: string, description: "Comma-separated tags" }
|
||||
requester_email: { type: string, description: "Requester email (optional)" }
|
||||
required: [subject, comment]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_ticket
|
||||
name: zendesk-get-ticket
|
||||
description: "Get a ticket by ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
ticket_id: { type: string, description: "Ticket ID" }
|
||||
required: [ticket_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: update_ticket
|
||||
name: zendesk-update-ticket
|
||||
description: "Update a ticket (status, priority, or assignee)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
ticket_id: { type: string, description: "Ticket ID" }
|
||||
status: { type: string, description: "new, open, pending, hold, solved, or closed" }
|
||||
priority: { type: string, description: "low, normal, high, or urgent" }
|
||||
assignee_id: { type: string, description: "Assignee user ID" }
|
||||
required: [ticket_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: search_tickets
|
||||
name: zendesk-search-tickets
|
||||
description: "Search tickets with a Zendesk search query."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
query: { type: string, description: "Search query (e.g. status:open priority:high)" }
|
||||
required: [query]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: add_comment
|
||||
name: zendesk-add-comment
|
||||
description: "Add a comment to a ticket (public or internal note)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
ticket_id: { type: string, description: "Ticket ID" }
|
||||
comment: { type: string, description: "Comment body" }
|
||||
public: { type: boolean, description: "Public comment (default true; false = internal note)" }
|
||||
required: [ticket_id, comment]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_users
|
||||
name: zendesk-list-users
|
||||
description: "List users (optionally filter by role)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
role: { type: string, description: "Optional role filter (end-user, agent, admin)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: zendesk-test-connection
|
||||
description: "Verify connectivity and credentials (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,67 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
ticket_id = inputs.get("ticket_id")
|
||||
if not ticket_id:
|
||||
raise Exception("ticket_id is required")
|
||||
|
||||
comment = str(inputs.get("comment", "")).strip()
|
||||
if not comment:
|
||||
raise Exception("comment is required")
|
||||
|
||||
public = inputs.get("public")
|
||||
public_val = True if public is None else bool(public)
|
||||
|
||||
body = {"ticket": {"comment": {"body": comment, "public": public_val}}}
|
||||
return request("PUT", "/tickets/" + q(ticket_id) + ".json", cfg, body=body)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,79 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
subject = str(inputs.get("subject", "")).strip()
|
||||
if not subject:
|
||||
raise Exception("subject is required")
|
||||
comment = str(inputs.get("comment", "")).strip()
|
||||
if not comment:
|
||||
raise Exception("comment is required")
|
||||
|
||||
ticket = {"subject": subject, "comment": {"body": comment}}
|
||||
|
||||
priority = inputs.get("priority")
|
||||
if priority:
|
||||
ticket["priority"] = priority
|
||||
|
||||
ticket_type = inputs.get("type")
|
||||
if ticket_type:
|
||||
ticket["type"] = ticket_type
|
||||
|
||||
tags = inputs.get("tags")
|
||||
if tags:
|
||||
tags_list = [s.strip() for s in str(tags).split(",") if s.strip()]
|
||||
if tags_list:
|
||||
ticket["tags"] = tags_list
|
||||
|
||||
requester_email = inputs.get("requester_email")
|
||||
if requester_email:
|
||||
ticket["requester"] = {"email": requester_email}
|
||||
|
||||
return request("POST", "/tickets.json", cfg, body={"ticket": ticket})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,59 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
ticket_id = inputs.get("ticket_id")
|
||||
if not ticket_id:
|
||||
raise Exception("ticket_id is required")
|
||||
|
||||
return request("GET", "/tickets/" + q(ticket_id) + ".json", cfg)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,53 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
role = inputs.get("role")
|
||||
return request("GET", "/users.json", cfg, params={"role": role})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,56 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
query = str(inputs.get("query", "")).strip()
|
||||
if not query:
|
||||
raise Exception("query is required")
|
||||
|
||||
return request("GET", "/search.json", cfg, params={"query": "type:ticket " + query})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,53 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
request("GET", "/users/me.json", cfg)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,76 @@
|
||||
import json, os, sys, base64, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _base(cfg):
|
||||
return "https://" + str(cfg.get("subdomain", "")) + ".zendesk.com/api/v2"
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
raw = str(cfg.get("email", "")) + "/token:" + str(cfg.get("api_token", ""))
|
||||
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8")
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = _base(cfg) + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": _auth(cfg), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
ticket_id = inputs.get("ticket_id")
|
||||
if not ticket_id:
|
||||
raise Exception("ticket_id is required")
|
||||
|
||||
ticket = {}
|
||||
|
||||
status = inputs.get("status")
|
||||
if status:
|
||||
ticket["status"] = status
|
||||
|
||||
priority = inputs.get("priority")
|
||||
if priority:
|
||||
ticket["priority"] = priority
|
||||
|
||||
assignee_id = inputs.get("assignee_id")
|
||||
if assignee_id not in (None, ""):
|
||||
ticket["assignee_id"] = int(assignee_id)
|
||||
|
||||
if not ticket:
|
||||
raise Exception("at least one field to update is required")
|
||||
|
||||
return request("PUT", "/tickets/" + q(ticket_id) + ".json", cfg, body={"ticket": ticket})
|
||||
|
||||
|
||||
_run(main)
|
||||
Reference in New Issue
Block a user