diff --git a/toolkits/zendesk/.pre-commit-config.yaml b/toolkits/zendesk/.pre-commit-config.yaml
new file mode 100644
index 00000000..6a345ba4
--- /dev/null
+++ b/toolkits/zendesk/.pre-commit-config.yaml
@@ -0,0 +1,18 @@
+files: ^.*/zendesk/.*
+repos:
+ - repo: https://github.com/pre-commit/pre-commit-hooks
+ rev: "v4.4.0"
+ hooks:
+ - id: check-case-conflict
+ - id: check-merge-conflict
+ - id: check-toml
+ - id: check-yaml
+ - id: end-of-file-fixer
+ - id: trailing-whitespace
+
+ - repo: https://github.com/astral-sh/ruff-pre-commit
+ rev: v0.6.7
+ hooks:
+ - id: ruff
+ args: [--fix]
+ - id: ruff-format
diff --git a/toolkits/zendesk/.ruff.toml b/toolkits/zendesk/.ruff.toml
new file mode 100644
index 00000000..2315c4aa
--- /dev/null
+++ b/toolkits/zendesk/.ruff.toml
@@ -0,0 +1,46 @@
+target-version = "py310"
+line-length = 100
+fix = true
+
+[lint]
+select = [
+ # flake8-2020
+ "YTT",
+ # flake8-bandit
+ "S",
+ # flake8-bugbear
+ "B",
+ # flake8-builtins
+ "A",
+ # flake8-comprehensions
+ "C4",
+ # flake8-debugger
+ "T10",
+ # flake8-simplify
+ "SIM",
+ # isort
+ "I",
+ # mccabe
+ "C90",
+ # pycodestyle
+ "E", "W",
+ # pyflakes
+ "F",
+ # pygrep-hooks
+ "PGH",
+ # pyupgrade
+ "UP",
+ # ruff
+ "RUF",
+ # tryceratops
+ "TRY",
+]
+
+ignore = ["C901"]
+
+[lint.per-file-ignores]
+"**/tests/*" = ["S101"]
+
+[format]
+preview = true
+skip-magic-trailing-comma = false
diff --git a/toolkits/zendesk/Makefile b/toolkits/zendesk/Makefile
new file mode 100644
index 00000000..0a8969be
--- /dev/null
+++ b/toolkits/zendesk/Makefile
@@ -0,0 +1,55 @@
+.PHONY: help
+
+help:
+ @echo "🛠️ github Commands:\n"
+ @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
+
+.PHONY: install
+install: ## Install the uv environment and install all packages with dependencies
+ @echo "🚀 Creating virtual environment and installing all packages using uv"
+ @uv sync --active --all-extras --no-sources
+ @if [ -f .pre-commit-config.yaml ]; then uv run --no-sources pre-commit install; fi
+ @echo "✅ All packages and dependencies installed via uv"
+
+.PHONY: install-local
+install-local: ## Install the uv environment and install all packages with dependencies with local Arcade sources
+ @echo "🚀 Creating virtual environment and installing all packages using uv"
+ @uv sync --active --all-extras
+ @if [ -f .pre-commit-config.yaml ]; then uv run pre-commit install; fi
+ @echo "✅ All packages and dependencies installed via uv"
+
+.PHONY: build
+build: clean-build ## Build wheel file using poetry
+ @echo "🚀 Creating wheel file"
+ uv build
+
+.PHONY: clean-build
+clean-build: ## clean build artifacts
+ @echo "🗑️ Cleaning dist directory"
+ rm -rf dist
+
+.PHONY: test
+test: ## Test the code with pytest
+ @echo "🚀 Testing code: Running pytest"
+ @uv run --no-sources pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
+
+.PHONY: coverage
+coverage: ## Generate coverage report
+ @echo "coverage report"
+ @uv run --no-sources coverage report
+ @echo "Generating coverage report"
+ @uv run --no-sources coverage html
+
+.PHONY: bump-version
+bump-version: ## Bump the version in the pyproject.toml file by a patch version
+ @echo "🚀 Bumping version in pyproject.toml"
+ uv version --no-sources --bump patch
+
+.PHONY: check
+check: ## Run code quality tools.
+ @if [ -f .pre-commit-config.yaml ]; then\
+ echo "🚀 Linting code: Running pre-commit";\
+ uv run --no-sources pre-commit run -a;\
+ fi
+ @echo "🚀 Static type checking: Running mypy"
+ @uv run --no-sources mypy --config-file=pyproject.toml
diff --git a/toolkits/zendesk/arcade_zendesk/__init__.py b/toolkits/zendesk/arcade_zendesk/__init__.py
new file mode 100644
index 00000000..373988de
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/__init__.py
@@ -0,0 +1,15 @@
+from arcade_zendesk.tools import (
+ add_ticket_comment,
+ get_ticket_comments,
+ list_tickets,
+ mark_ticket_solved,
+ search_articles,
+)
+
+__all__ = [
+ "list_tickets",
+ "add_ticket_comment",
+ "get_ticket_comments",
+ "mark_ticket_solved",
+ "search_articles",
+]
diff --git a/toolkits/zendesk/arcade_zendesk/enums.py b/toolkits/zendesk/arcade_zendesk/enums.py
new file mode 100644
index 00000000..f135bf44
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/enums.py
@@ -0,0 +1,27 @@
+"""Enums for the Zendesk toolkit."""
+
+from enum import Enum
+
+
+class ArticleSortBy(Enum):
+ """Sort fields for article search results."""
+
+ CREATED_AT = "created_at"
+ RELEVANCE = "relevance"
+
+
+class SortOrder(Enum):
+ """Sort order direction."""
+
+ ASC = "asc"
+ DESC = "desc"
+
+
+class TicketStatus(Enum):
+ """Valid ticket statuses."""
+
+ NEW = "new"
+ OPEN = "open"
+ PENDING = "pending"
+ SOLVED = "solved"
+ CLOSED = "closed"
diff --git a/toolkits/zendesk/arcade_zendesk/tools/__init__.py b/toolkits/zendesk/arcade_zendesk/tools/__init__.py
new file mode 100644
index 00000000..99699a9b
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/tools/__init__.py
@@ -0,0 +1,15 @@
+from arcade_zendesk.tools.search_articles import search_articles
+from arcade_zendesk.tools.tickets import (
+ add_ticket_comment,
+ get_ticket_comments,
+ list_tickets,
+ mark_ticket_solved,
+)
+
+__all__ = [
+ "list_tickets",
+ "add_ticket_comment",
+ "get_ticket_comments",
+ "mark_ticket_solved",
+ "search_articles",
+]
diff --git a/toolkits/zendesk/arcade_zendesk/tools/search_articles.py b/toolkits/zendesk/arcade_zendesk/tools/search_articles.py
new file mode 100644
index 00000000..b47115be
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/tools/search_articles.py
@@ -0,0 +1,227 @@
+import logging
+from typing import Annotated, Any
+
+import httpx
+from arcade_tdk import ToolContext, tool
+from arcade_tdk.auth import OAuth2
+from arcade_tdk.errors import RetryableToolError, ToolExecutionError
+
+from arcade_zendesk.enums import ArticleSortBy, SortOrder
+from arcade_zendesk.utils import (
+ fetch_paginated_results,
+ get_zendesk_subdomain,
+ process_search_results,
+ validate_date_format,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@tool(
+ requires_auth=OAuth2(id="zendesk", scopes=["read"]),
+ requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def search_articles(
+ context: ToolContext,
+ query: Annotated[
+ str | None,
+ "Search text to match against articles. Supports quoted expressions for exact matching",
+ ] = None,
+ label_names: Annotated[
+ list[str] | None,
+ "List of label names to filter by (case-insensitive). Article must have at least "
+ "one matching label. Available on Professional/Enterprise plans only",
+ ] = None,
+ created_after: Annotated[
+ str | None,
+ "Filter articles created after this date (format: YYYY-MM-DD)",
+ ] = None,
+ created_before: Annotated[
+ str | None,
+ "Filter articles created before this date (format: YYYY-MM-DD)",
+ ] = None,
+ created_at: Annotated[
+ str | None,
+ "Filter articles created on this exact date (format: YYYY-MM-DD)",
+ ] = None,
+ sort_by: Annotated[
+ ArticleSortBy | None,
+ "Field to sort articles by. Defaults to relevance according to the search query",
+ ] = None,
+ sort_order: Annotated[
+ SortOrder | None,
+ "Sort order direction. Defaults to descending",
+ ] = None,
+ limit: Annotated[
+ int,
+ "Number of articles to return. Defaults to 30",
+ ] = 30,
+ offset: Annotated[
+ int,
+ "Number of articles to skip before returning results. Defaults to 0",
+ ] = 0,
+ include_body: Annotated[
+ bool,
+ "Include article body content in results. Bodies will be cleaned of HTML and truncated",
+ ] = True,
+ max_article_length: Annotated[
+ int | None,
+ "Maximum length for article body content in characters. "
+ "Set to None for no limit. Defaults to 500",
+ ] = 500,
+) -> Annotated[
+ dict[str, Any],
+ "Article search results with pagination metadata. Includes 'next_offset' when more "
+ "results are available. Simply use this value as the 'offset' parameter in your next "
+ "call to fetch the next batch",
+]:
+ """
+ Search for Help Center articles in your Zendesk knowledge base.
+
+ This tool searches specifically for published knowledge base articles that provide
+ solutions and guidance to users. At least one search parameter (query or label_names)
+ must be provided.
+
+ PAGINATION:
+ - The response includes 'next_offset' when more results are available
+ - To fetch the next batch, simply pass the 'next_offset' value as the 'offset' parameter
+ - If 'next_offset' is not present, you've reached the end of available results
+ - The tool automatically handles fetching from the correct page based on your offset
+
+ IMPORTANT: ALL FILTERS CAN BE COMBINED IN A SINGLE CALL
+ You can combine multiple filters (query, labels, dates) in one search request.
+ Do NOT make separate tool calls - combine all relevant filters together.
+ """
+
+ # Validate date parameters
+ date_params = {
+ "created_after": created_after,
+ "created_before": created_before,
+ "created_at": created_at,
+ }
+
+ for param_name, param_value in date_params.items():
+ if param_value and not validate_date_format(param_value):
+ raise RetryableToolError(
+ message=(
+ f"Invalid date format for {param_name}: '{param_value}'. "
+ "Please use YYYY-MM-DD format."
+ ),
+ developer_message=(
+ f"Date validation failed for parameter '{param_name}' "
+ f"with value '{param_value}'"
+ ),
+ retry_after_ms=500,
+ additional_prompt_content="Use format YYYY-MM-DD.",
+ )
+
+ # Validate limit and offset parameters
+ if limit < 1:
+ raise RetryableToolError(
+ message="limit must be at least 1.",
+ developer_message=f"Invalid limit value: {limit}",
+ retry_after_ms=100,
+ additional_prompt_content="Provide a positive limit value",
+ )
+
+ if offset < 0:
+ raise RetryableToolError(
+ message="offset cannot be negative.",
+ developer_message=f"Invalid offset value: {offset}",
+ retry_after_ms=100,
+ additional_prompt_content="Provide a non-negative offset value",
+ )
+
+ # Validate that at least one search parameter is provided
+ if not any([query, label_names]):
+ raise RetryableToolError(
+ message="At least one search parameter must be provided.",
+ developer_message="No search parameters were provided",
+ retry_after_ms=100,
+ additional_prompt_content=(
+ "Provide at least one of: query text or a list of label names"
+ ),
+ )
+
+ auth_token = context.get_auth_token_or_empty()
+ subdomain = get_zendesk_subdomain(context)
+
+ url = f"https://{subdomain}.zendesk.com/api/v2/help_center/articles/search"
+
+ # Base parameters for the search
+ base_params: dict[str, Any] = {
+ "per_page": 100, # Max allowed per page
+ }
+
+ if query:
+ base_params["query"] = query
+
+ if label_names:
+ base_params["label_names"] = ",".join(label_names)
+
+ if created_after:
+ base_params["created_after"] = created_after
+
+ if created_before:
+ base_params["created_before"] = created_before
+
+ if created_at:
+ base_params["created_at"] = created_at
+
+ if sort_by:
+ base_params["sort_by"] = sort_by.value
+
+ if sort_order:
+ base_params["sort_order"] = sort_order.value
+
+ async with httpx.AsyncClient() as client:
+ try:
+ headers = {
+ "Authorization": f"Bearer {auth_token}",
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ }
+
+ data = await fetch_paginated_results(
+ client=client,
+ url=url,
+ headers=headers,
+ params=base_params,
+ offset=offset,
+ limit=limit,
+ )
+
+ if "results" in data:
+ data["results"] = process_search_results(
+ data["results"], include_body=include_body, max_body_length=max_article_length
+ )
+
+ logger.info(f"Article search returned {data.get('count', 0)} results")
+
+ except httpx.HTTPStatusError as e:
+ logger.exception(f"HTTP error during article search: {e.response.status_code}")
+ raise ToolExecutionError(
+ message=f"Failed to search articles: HTTP {e.response.status_code}",
+ developer_message=(
+ f"HTTP {e.response.status_code} error: {e.response.text}. "
+ f"URL: {url}, base_params: {base_params}"
+ ),
+ ) from e
+ except httpx.TimeoutException as e:
+ logger.exception("Timeout during article search")
+ raise RetryableToolError(
+ message="Request timed out while searching articles.",
+ developer_message=f"Timeout occurred. URL: {url}, base_params: {base_params}",
+ retry_after_ms=5000,
+ ) from e
+ except Exception as e:
+ logger.exception("Unexpected error during article search")
+ raise ToolExecutionError(
+ message=f"Failed to search articles: {e!s}",
+ developer_message=(
+ f"Unexpected error: {type(e).__name__}: {e!s}. "
+ f"URL: {url}, base_params: {base_params}"
+ ),
+ ) from e
+ else:
+ return data
diff --git a/toolkits/zendesk/arcade_zendesk/tools/tickets.py b/toolkits/zendesk/arcade_zendesk/tools/tickets.py
new file mode 100644
index 00000000..814917f0
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/tools/tickets.py
@@ -0,0 +1,414 @@
+from typing import Annotated, Any
+
+import httpx
+from arcade_tdk import ToolContext, tool
+from arcade_tdk.auth import OAuth2
+from arcade_tdk.errors import RetryableToolError, ToolExecutionError
+
+from arcade_zendesk.enums import SortOrder, TicketStatus
+from arcade_zendesk.utils import fetch_paginated_results, get_zendesk_subdomain
+
+
+def _handle_ticket_not_found(response: httpx.Response, ticket_id: int) -> None:
+ """Handle 404 responses for ticket operations."""
+ if response.status_code == 404:
+ raise RetryableToolError(
+ message=f"Ticket #{ticket_id} not found.",
+ developer_message=f"Ticket with ID {ticket_id} does not exist",
+ retry_after_ms=500,
+ additional_prompt_content="Please verify the ticket ID and try again",
+ )
+
+
+@tool(
+ requires_auth=OAuth2(id="zendesk", scopes=["read"]),
+ requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def list_tickets(
+ context: ToolContext,
+ status: Annotated[
+ TicketStatus,
+ "The status of tickets to filter by. Defaults to 'open'",
+ ] = TicketStatus.OPEN,
+ limit: Annotated[
+ int,
+ "Number of tickets to return. Defaults to 30",
+ ] = 30,
+ offset: Annotated[
+ int,
+ "Number of tickets to skip before returning results. Defaults to 0",
+ ] = 0,
+ sort_order: Annotated[
+ SortOrder,
+ "Sort order for tickets by ID. 'asc' returns oldest first, 'desc' returns newest first. "
+ "Defaults to 'desc'",
+ ] = SortOrder.DESC,
+) -> Annotated[
+ dict[str, Any],
+ "A dictionary containing tickets list (each with html_url), count, and pagination metadata. "
+ "Includes 'next_offset' when more results are available",
+]:
+ """List tickets from your Zendesk account with offset-based pagination.
+
+ By default, returns tickets sorted by ID with newest tickets first (desc).
+
+ Each ticket in the response includes an 'html_url' field with the direct link
+ to view the ticket in Zendesk.
+
+ PAGINATION:
+ - The response includes 'next_offset' when more results are available
+ - To fetch the next batch, simply pass the 'next_offset' value as the 'offset' parameter
+ - If 'next_offset' is not present, you've reached the end of available results
+ """
+
+ # Validate limit and offset parameters
+ if limit < 1:
+ raise RetryableToolError(
+ message="limit must be at least 1.",
+ developer_message=f"Invalid limit value: {limit}",
+ retry_after_ms=100,
+ additional_prompt_content="Provide a positive limit value",
+ )
+
+ if offset < 0:
+ raise RetryableToolError(
+ message="offset cannot be negative.",
+ developer_message=f"Invalid offset value: {offset}",
+ retry_after_ms=100,
+ additional_prompt_content="Provide a non-negative offset value",
+ )
+
+ # Get the authorization token
+ token = context.get_auth_token_or_empty()
+ subdomain = get_zendesk_subdomain(context)
+
+ # Build the API URL
+ url = f"https://{subdomain}.zendesk.com/api/v2/tickets.json"
+
+ # Base parameters for the request
+ base_params: dict[str, Any] = {
+ "status": status.value,
+ "per_page": 100, # Max allowed per page
+ "sort_order": sort_order.value,
+ }
+
+ # Make the API request
+ async with httpx.AsyncClient() as client:
+ try:
+ headers = {
+ "Authorization": f"Bearer {token}",
+ "Content-Type": "application/json",
+ }
+
+ # Use the fetch_paginated_results utility
+ data = await fetch_paginated_results(
+ client=client,
+ url=url,
+ headers=headers,
+ params=base_params,
+ offset=offset,
+ limit=limit,
+ )
+
+ # Process tickets to add html_url and remove api url
+ tickets = data.get("results", [])
+ for ticket in tickets:
+ if "id" in ticket:
+ ticket["html_url"] = (
+ f"https://{subdomain}.zendesk.com/agent/tickets/{ticket['id']}"
+ )
+ # Remove API url to avoid confusion
+ if "url" in ticket:
+ del ticket["url"]
+
+ # Build the result with consistent structure
+ result = {
+ "tickets": tickets,
+ "count": data.get("count", len(tickets)),
+ }
+
+ # Add next_offset if present
+ if "next_offset" in data:
+ result["next_offset"] = data["next_offset"]
+
+ except httpx.HTTPStatusError as e:
+ raise ToolExecutionError(
+ message=f"Failed to list tickets: HTTP {e.response.status_code}",
+ developer_message=(
+ f"HTTP {e.response.status_code} error: {e.response.text}. "
+ f"URL: {url}, params: {base_params}"
+ ),
+ ) from e
+ except httpx.TimeoutException as e:
+ raise RetryableToolError(
+ message="Request timed out while listing tickets.",
+ developer_message=f"Timeout occurred. URL: {url}, params: {base_params}",
+ retry_after_ms=5000,
+ additional_prompt_content="Try reducing limit or using more specific filters.",
+ ) from e
+ except Exception as e:
+ raise ToolExecutionError(
+ message=f"Failed to list tickets: {e!s}",
+ developer_message=(
+ f"Unexpected error: {type(e).__name__}: {e!s}. "
+ f"URL: {url}, params: {base_params}"
+ ),
+ ) from e
+ else:
+ return result
+
+
+@tool(
+ requires_auth=OAuth2(id="zendesk", scopes=["read"]),
+ requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def get_ticket_comments(
+ context: ToolContext,
+ ticket_id: Annotated[int, "The ID of the ticket to get comments for"],
+) -> Annotated[
+ dict[str, Any], "A dictionary containing the ticket comments, metadata, and ticket URL"
+]:
+ """Get all comments for a specific Zendesk ticket, including the original description.
+
+ The first comment is always the ticket's original description/content.
+ Subsequent comments show the conversation history.
+
+ Each comment includes:
+ - author_id: ID of the comment author
+ - body: The comment text
+ - created_at: Timestamp when comment was created
+ - public: Whether the comment is public or internal
+ - attachments: List of file attachments (if any) with file_name, content_url, size, etc.
+ """
+
+ # Get the authorization token
+ token = context.get_auth_token_or_empty()
+ subdomain = get_zendesk_subdomain(context)
+
+ # Zendesk API endpoint for ticket comments
+ url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}/comments.json"
+
+ # Make the API request
+ async with httpx.AsyncClient() as client:
+ try:
+ headers = {
+ "Authorization": f"Bearer {token}",
+ "Content-Type": "application/json",
+ }
+
+ response = await client.get(url, headers=headers)
+ _handle_ticket_not_found(response, ticket_id)
+ response.raise_for_status()
+
+ data = response.json()
+ comments = data.get("comments", [])
+
+ return {
+ "ticket_id": ticket_id,
+ "comments": comments,
+ "count": len(comments),
+ }
+
+ except RetryableToolError:
+ # Re-raise our custom errors
+ raise
+ except httpx.HTTPStatusError as e:
+ raise ToolExecutionError(
+ message=f"Failed to get ticket comments: HTTP {e.response.status_code}",
+ developer_message=(
+ f"HTTP {e.response.status_code} error: {e.response.text}. URL: {url}"
+ ),
+ ) from e
+ except httpx.TimeoutException as e:
+ raise RetryableToolError(
+ message="Request timed out while getting ticket comments.",
+ developer_message=f"Timeout occurred. URL: {url}",
+ retry_after_ms=5000,
+ additional_prompt_content="Try again in a few moments.",
+ ) from e
+ except Exception as e:
+ raise ToolExecutionError(
+ message=f"Failed to get ticket comments: {e!s}",
+ developer_message=f"Unexpected error: {type(e).__name__}: {e!s}. URL: {url}",
+ ) from e
+
+
+@tool(
+ requires_auth=OAuth2(id="zendesk", scopes=["tickets:write"]),
+ requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def add_ticket_comment(
+ context: ToolContext,
+ ticket_id: Annotated[int, "The ID of the ticket to comment on"],
+ comment_body: Annotated[str, "The text of the comment"],
+ public: Annotated[
+ bool, "Whether the comment is public (visible to requester) or internal. Defaults to True"
+ ] = True,
+) -> Annotated[
+ dict[str, Any], "A dictionary containing the result of the comment operation and ticket URL"
+]:
+ """Add a comment to an existing Zendesk ticket.
+
+ The returned ticket object includes an 'html_url' field with the direct link
+ to view the ticket in Zendesk.
+ """
+
+ # Get the authorization token
+ token = context.get_auth_token_or_empty()
+ subdomain = get_zendesk_subdomain(context)
+
+ # Zendesk API endpoint for updating ticket
+ url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
+
+ # Prepare the request body
+ request_body = {"ticket": {"comment": {"body": comment_body, "public": public}}}
+
+ # Make the API request
+ async with httpx.AsyncClient() as client:
+ try:
+ headers = {
+ "Authorization": f"Bearer {token}",
+ "Content-Type": "application/json",
+ }
+
+ response = await client.put(url, headers=headers, json=request_body)
+ _handle_ticket_not_found(response, ticket_id)
+ response.raise_for_status()
+
+ data = response.json()
+ ticket = data.get("ticket", {})
+
+ # Add web interface URL if not present
+ if "id" in ticket and "html_url" not in ticket:
+ ticket["html_url"] = f"https://{subdomain}.zendesk.com/agent/tickets/{ticket['id']}"
+ # Remove API url to avoid confusion
+ if "url" in ticket:
+ del ticket["url"]
+
+ except RetryableToolError:
+ # Re-raise our custom errors
+ raise
+ except httpx.HTTPStatusError as e:
+ raise ToolExecutionError(
+ message=f"Failed to add ticket comment: HTTP {e.response.status_code}",
+ developer_message=(
+ f"HTTP {e.response.status_code} error: {e.response.text}. "
+ f"URL: {url}, body: {request_body}"
+ ),
+ ) from e
+ except httpx.TimeoutException as e:
+ raise RetryableToolError(
+ message="Request timed out while adding ticket comment.",
+ developer_message=f"Timeout occurred. URL: {url}",
+ retry_after_ms=5000,
+ additional_prompt_content="Try again in a few moments.",
+ ) from e
+ except Exception as e:
+ raise ToolExecutionError(
+ message=f"Failed to add ticket comment: {e!s}",
+ developer_message=f"Unexpected error: {type(e).__name__}: {e!s}. URL: {url}",
+ ) from e
+ else:
+ return {
+ "success": True,
+ "ticket_id": ticket_id,
+ "comment_type": "public" if public else "internal",
+ "ticket": ticket,
+ }
+
+
+@tool(
+ requires_auth=OAuth2(id="zendesk", scopes=["tickets:write"]),
+ requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def mark_ticket_solved(
+ context: ToolContext,
+ ticket_id: Annotated[int, "The ID of the ticket to mark as solved"],
+ comment_body: Annotated[
+ str | None,
+ "Optional final comment to add when solving the ticket",
+ ] = None,
+ comment_public: Annotated[
+ bool, "Whether the comment is visible to the requester. Defaults to False"
+ ] = False,
+) -> Annotated[dict[str, Any], "A dictionary containing the result of the solve operation"]:
+ """Mark a Zendesk ticket as solved, optionally with a final comment.
+
+ The returned ticket object includes an 'html_url' field with the direct link
+ to view the ticket in Zendesk.
+ """
+
+ # Get the authorization token
+ token = context.get_auth_token_or_empty()
+ subdomain = get_zendesk_subdomain(context)
+
+ # Zendesk API endpoint for updating ticket
+ url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
+
+ # Prepare the request body
+ request_body: dict[str, Any] = {"ticket": {"status": "solved"}}
+
+ # Add resolution comment if provided
+ if comment_body:
+ request_body["ticket"]["comment"] = {
+ "body": comment_body,
+ "public": comment_public,
+ }
+
+ # Make the API request
+ async with httpx.AsyncClient() as client:
+ try:
+ headers = {
+ "Authorization": f"Bearer {token}",
+ "Content-Type": "application/json",
+ }
+
+ response = await client.put(url, headers=headers, json=request_body)
+ _handle_ticket_not_found(response, ticket_id)
+ response.raise_for_status()
+
+ data = response.json()
+ ticket = data.get("ticket", {})
+
+ # Add web interface URL if not present
+ if "id" in ticket and "html_url" not in ticket:
+ ticket["html_url"] = f"https://{subdomain}.zendesk.com/agent/tickets/{ticket['id']}"
+ # Remove API url to avoid confusion
+ if "url" in ticket:
+ del ticket["url"]
+
+ result = {
+ "success": True,
+ "ticket_id": ticket_id,
+ "status": "solved",
+ "ticket": ticket,
+ }
+ if comment_body:
+ result["comment_added"] = True
+ result["comment_type"] = "public" if comment_public else "internal"
+
+ except RetryableToolError:
+ # Re-raise our custom errors
+ raise
+ except httpx.HTTPStatusError as e:
+ raise ToolExecutionError(
+ message=f"Failed to mark ticket as solved: HTTP {e.response.status_code}",
+ developer_message=(
+ f"HTTP {e.response.status_code} error: {e.response.text}. "
+ f"URL: {url}, body: {request_body}"
+ ),
+ ) from e
+ except httpx.TimeoutException as e:
+ raise RetryableToolError(
+ message="Request timed out while marking ticket as solved.",
+ developer_message=f"Timeout occurred. URL: {url}",
+ retry_after_ms=5000,
+ additional_prompt_content="Try again in a few moments.",
+ ) from e
+ except Exception as e:
+ raise ToolExecutionError(
+ message=f"Failed to mark ticket as solved: {e!s}",
+ developer_message=f"Unexpected error: {type(e).__name__}: {e!s}. URL: {url}",
+ ) from e
+ else:
+ return result
diff --git a/toolkits/zendesk/arcade_zendesk/utils.py b/toolkits/zendesk/arcade_zendesk/utils.py
new file mode 100644
index 00000000..9ac405a3
--- /dev/null
+++ b/toolkits/zendesk/arcade_zendesk/utils.py
@@ -0,0 +1,216 @@
+import logging
+import re
+from typing import Any
+
+import httpx
+from arcade_tdk import ToolContext
+from arcade_tdk.errors import ToolExecutionError
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MAX_BODY_LENGTH = 500 # Default max length for article body content
+
+
+async def fetch_paginated_results(
+ client: httpx.AsyncClient,
+ url: str,
+ headers: dict[str, str],
+ params: dict[str, Any],
+ offset: int,
+ limit: int,
+) -> dict[str, Any]:
+ """
+ Fetch paginated results using offset and limit pattern.
+
+ This function internally manages pagination to fulfill the requested offset and limit,
+ fetching multiple pages as needed.
+
+ Args:
+ client: The HTTP client to use
+ url: The API endpoint URL
+ headers: Request headers including authorization
+ params: Base query parameters (without pagination params)
+ offset: Number of items to skip
+ limit: Number of items to return
+
+ Returns:
+ Dict containing:
+ - results: List of fetched items
+ - count: Number of items returned
+ - next_offset: Present only if more results are available
+ """
+ # Calculate pagination parameters
+ # Most Zendesk APIs use 1-based page numbering
+ items_per_page = params.get("per_page", 100) # Use per_page from params or default to 100
+ start_page = (offset // items_per_page) + 1
+ start_index = offset % items_per_page
+
+ # Collect results across multiple pages if needed
+ all_results = []
+ current_page = start_page
+ items_collected = 0
+ has_more = False
+ last_page_had_more_items = False
+
+ while items_collected < limit:
+ # Set the current page
+ page_params = params.copy()
+ page_params["page"] = current_page
+
+ response = await client.get(url, headers=headers, params=page_params, timeout=30.0)
+ response.raise_for_status()
+ page_data = response.json()
+
+ # Extract results from current page (handle both "results" and "tickets" keys)
+ page_results = page_data.get("results", page_data.get("tickets", []))
+
+ # If this is the first page, skip to the start index
+ if current_page == start_page:
+ page_results = page_results[start_index:]
+
+ # Take only what we need to reach the limit
+ items_needed = limit - items_collected
+ results_to_add = page_results[:items_needed]
+ all_results.extend(results_to_add)
+ items_collected += len(results_to_add)
+
+ # Check if we left items on this page
+ if len(page_results) > items_needed:
+ last_page_had_more_items = True
+
+ # Check if there are more pages
+ has_more = page_data.get("next_page") is not None
+
+ # Stop if we've collected enough or no more pages
+ if items_collected >= limit or not has_more:
+ break
+
+ current_page += 1
+
+ # Build the response
+ result = {
+ "results": all_results,
+ "count": len(all_results),
+ }
+
+ # Add next_offset if there might be more results
+ # This happens when:
+ # 1. We got exactly the limit requested AND (there are more pages OR we left items on the page)
+ # 2. We didn't get the full limit but there are more pages available
+ if (len(all_results) == limit and (has_more or last_page_had_more_items)) or (
+ len(all_results) < limit and has_more
+ ):
+ result["next_offset"] = offset + len(all_results)
+
+ return result
+
+
+def clean_html_text(text: str | None) -> str:
+ """Remove HTML tags and clean up text."""
+ if not text:
+ return ""
+
+ soup = BeautifulSoup(text, "html.parser")
+ clean_text: str = soup.get_text(separator=" ")
+
+ clean_text = re.sub(r"\n+", "\n", clean_text)
+
+ clean_text = re.sub(r"\s+", " ", clean_text)
+
+ clean_text = "\n".join(line.strip() for line in clean_text.split("\n"))
+
+ return clean_text.strip()
+
+
+def truncate_text(
+ text: str | None, max_length: int, suffix: str = " ... [truncated]"
+) -> str | None:
+ """Truncate text to a maximum length with a suffix."""
+ if not text or len(text) <= max_length:
+ return text
+
+ truncate_at = max_length - len(suffix)
+ if truncate_at <= 0:
+ return suffix
+
+ return text[:truncate_at] + suffix
+
+
+def process_article_body(body: str | None, max_length: int | None = None) -> str | None:
+ """Process article body by cleaning HTML and optionally truncating."""
+ if not body:
+ return None
+
+ cleaned_text: str = clean_html_text(body)
+
+ if max_length and len(cleaned_text) > max_length:
+ result = truncate_text(cleaned_text, max_length)
+ return result
+
+ return cleaned_text
+
+
+def process_search_results(
+ results: list[dict[str, Any]],
+ include_body: bool = False,
+ max_body_length: int | None = DEFAULT_MAX_BODY_LENGTH,
+) -> list[dict[str, Any]]:
+ """Process search results to clean up data and restructure with content and metadata."""
+ processed_results = []
+
+ for result in results:
+ body_content = result.get("body", "")
+ cleaned_content = None
+
+ if include_body and body_content:
+ cleaned_content = process_article_body(body_content, max_body_length)
+
+ processed_result: dict[str, Any] = {"content": cleaned_content, "metadata": {}}
+
+ for key, value in result.items():
+ if key != "body":
+ processed_result["metadata"][key] = value
+
+ processed_results.append(processed_result)
+
+ return processed_results
+
+
+def validate_date_format(date_string: str) -> bool:
+ """Validate that a date string matches YYYY-MM-DD format and is a valid date."""
+ from datetime import datetime
+
+ try:
+ parsed_date = datetime.strptime(date_string, "%Y-%m-%d")
+ # Ensure the input matches the expected format exactly
+ return parsed_date.strftime("%Y-%m-%d") == date_string
+ except ValueError:
+ return False
+
+
+def get_zendesk_subdomain(context: ToolContext) -> str:
+ """
+ Get the Zendesk subdomain from secrets with proper error handling.
+
+ Args:
+ context: The tool context containing secrets
+
+ Returns:
+ The Zendesk subdomain
+
+ Raises:
+ ToolExecutionError: If the subdomain secret is not configured
+ """
+ try:
+ subdomain = context.get_secret("ZENDESK_SUBDOMAIN")
+ except ValueError:
+ raise ToolExecutionError(
+ message="Zendesk subdomain is not set.",
+ developer_message=(
+ "Zendesk subdomain is not set. Make sure to set the "
+ "'ZENDESK_SUBDOMAIN' secret in the Arcade Dashboard."
+ ),
+ ) from None
+ else:
+ return subdomain
diff --git a/toolkits/zendesk/evals/eval_articles.py b/toolkits/zendesk/evals/eval_articles.py
new file mode 100644
index 00000000..c96d84d6
--- /dev/null
+++ b/toolkits/zendesk/evals/eval_articles.py
@@ -0,0 +1,360 @@
+from datetime import timedelta
+
+from arcade_evals import (
+ DatetimeCritic,
+ EvalRubric,
+ EvalSuite,
+ ExpectedToolCall,
+ tool_eval,
+)
+from arcade_evals.critic import BinaryCritic, SimilarityCritic
+from arcade_tdk import ToolCatalog
+
+import arcade_zendesk
+from arcade_zendesk.enums import ArticleSortBy, SortOrder
+from arcade_zendesk.tools.search_articles import search_articles
+
+# Evaluation rubric
+rubric = EvalRubric(
+ fail_threshold=0.85,
+ warn_threshold=0.95,
+)
+
+
+catalog = ToolCatalog()
+catalog.add_module(arcade_zendesk)
+
+
+@tool_eval()
+def zendesk_search_articles_eval_suite() -> EvalSuite:
+ suite = EvalSuite(
+ name="Zendesk Search Articles Evaluation",
+ system_message=(
+ "You are an AI assistant with access to Zendesk Search Articles tool. "
+ "Use it to help users search for knowledge base articles and documentation."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Basic search scenarios
+ suite.add_case(
+ name="Basic search with query only",
+ user_message="Find articles about password reset",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": "password reset",
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=1.0),
+ ],
+ )
+
+ suite.add_case(
+ name="Search with specific result count",
+ user_message="Show me 25 articles about API documentation",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "API documentation", "limit": 25},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.7),
+ BinaryCritic(critic_field="limit", weight=0.3),
+ ],
+ )
+
+ # Date filtering scenarios
+ suite.add_case(
+ name="Search with created after date filter",
+ user_message="Find articles about security updates created after January 15, 2024",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "security updates", "created_after": "2024-01-15"},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.6),
+ DatetimeCritic(critic_field="created_after", weight=0.4, tolerance=timedelta(days=1)),
+ ],
+ )
+
+ suite.add_case(
+ name="Search with date range filter",
+ user_message="Show me articles about new features created between January and June 2024",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": "new features",
+ "created_after": "2024-01-01",
+ "created_before": "2024-06-30",
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.4),
+ DatetimeCritic(critic_field="created_after", weight=0.3, tolerance=timedelta(days=1)),
+ DatetimeCritic(critic_field="created_before", weight=0.3, tolerance=timedelta(days=1)),
+ ],
+ )
+
+ # Label filtering (Professional/Enterprise)
+ suite.add_case(
+ name="Search by labels only",
+ user_message="Show me articles tagged with windows and setup labels",
+ expected_tool_calls=[
+ ExpectedToolCall(func=search_articles, args={"label_names": ["windows", "setup"]})
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="label_names", weight=1.0),
+ ],
+ )
+
+ suite.add_case(
+ name="Search with query and labels",
+ user_message="Find installation guides with labels: macos, quickstart",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "installation guide", "label_names": ["macos", "quickstart"]},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.5),
+ SimilarityCritic(critic_field="label_names", weight=0.5),
+ ],
+ )
+
+ # Sorting scenarios
+ suite.add_case(
+ name="Search sorted by creation date ascending",
+ user_message="Find onboarding articles sorted by oldest first",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": "onboarding",
+ "sort_by": ArticleSortBy.CREATED_AT,
+ "sort_order": SortOrder.ASC,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.4),
+ BinaryCritic(critic_field="sort_by", weight=0.3),
+ BinaryCritic(critic_field="sort_order", weight=0.3),
+ ],
+ )
+
+ suite.add_case(
+ name="Search sorted by most recently created",
+ user_message="Show me troubleshooting guides sorted by latest creation",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": "troubleshooting guide",
+ "sort_by": ArticleSortBy.CREATED_AT,
+ "sort_order": SortOrder.DESC,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.4),
+ BinaryCritic(critic_field="sort_by", weight=0.3),
+ BinaryCritic(critic_field="sort_order", weight=0.3),
+ ],
+ )
+
+ # Pagination scenarios
+ suite.add_case(
+ name="Search with higher limit",
+ user_message="Show me 100 installation guides",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "installation guide", "limit": 100},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.7),
+ BinaryCritic(critic_field="limit", weight=0.3),
+ ],
+ )
+
+ suite.add_case(
+ name="Search with offset pagination",
+ user_message="Find API documentation, skip the first 50 results and show me the next 50",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "API documentation", "offset": 50, "limit": 50},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.4),
+ BinaryCritic(critic_field="offset", weight=0.3),
+ BinaryCritic(critic_field="limit", weight=0.3),
+ ],
+ )
+
+ # Complex search scenarios
+ suite.add_case(
+ name="Complex search with multiple filters",
+ user_message="Find recent troubleshooting articles about login issues "
+ "created after March 31, 2024, sorted by newest first",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": "login issues troubleshooting",
+ "created_after": "2024-03-31",
+ "sort_by": ArticleSortBy.CREATED_AT,
+ "sort_order": SortOrder.DESC,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.4),
+ DatetimeCritic(critic_field="created_after", weight=0.3, tolerance=timedelta(days=1)),
+ BinaryCritic(critic_field="sort_by", weight=0.15),
+ BinaryCritic(critic_field="sort_order", weight=0.15),
+ ],
+ )
+
+ # Content control
+ suite.add_case(
+ name="Search without article body content",
+ user_message="List article titles about billing without the full content",
+ expected_tool_calls=[
+ ExpectedToolCall(func=search_articles, args={"query": "billing", "include_body": False})
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.7),
+ BinaryCritic(critic_field="include_body", weight=0.3),
+ ],
+ )
+
+ # Edge cases
+ suite.add_case(
+ name="Search with exact phrase matching",
+ user_message='Find articles with the exact phrase "password reset procedure"',
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={
+ "query": '"password reset procedure"',
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="query", weight=1.0),
+ ],
+ )
+
+ return suite
+
+
+@tool_eval()
+def zendesk_search_articles_pagination_eval_suite() -> EvalSuite:
+ """Separate suite for pagination scenarios with context."""
+ suite = EvalSuite(
+ name="Zendesk Pagination Evaluation",
+ system_message=(
+ "You are an AI assistant with access to Zendesk Help Center tools. "
+ "Use them to help users search for knowledge base articles. "
+ "When users ask for more results, use appropriate pagination parameters."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Pagination with context
+ suite.add_case(
+ name="Initial search with pagination context",
+ user_message="I need to find all troubleshooting articles. "
+ "Start by showing me the first 20.",
+ expected_tool_calls=[
+ ExpectedToolCall(func=search_articles, args={"query": "troubleshooting", "limit": 20})
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.6),
+ BinaryCritic(critic_field="limit", weight=0.4),
+ ],
+ )
+
+ suite.add_case(
+ name="Request for more results after initial search",
+ user_message="Show me the next 20 troubleshooting articles",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=search_articles,
+ args={"query": "troubleshooting", "offset": 20, "limit": 20},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ SimilarityCritic(critic_field="query", weight=0.5),
+ BinaryCritic(critic_field="offset", weight=0.25),
+ BinaryCritic(critic_field="limit", weight=0.25),
+ ],
+ additional_messages=[
+ {
+ "role": "user",
+ "content": "I need to find all troubleshooting articles. "
+ "Start by showing me the first 20.",
+ },
+ {
+ "role": "assistant",
+ "content": "I'll search for troubleshooting articles and "
+ "show you the first 20 results.",
+ "tool_calls": [
+ {
+ "id": "call_1",
+ "type": "function",
+ "function": {
+ "name": "search_articles",
+ "arguments": '{"query": "troubleshooting", "limit": 20}',
+ },
+ }
+ ],
+ },
+ {
+ "role": "tool",
+ "content": '{"results": [{"content": "Troubleshooting guide 1", '
+ '"metadata": {"id": 1, "title": "How to troubleshoot login issues"}}], '
+ '"count": 20, "next_offset": 20}',
+ "tool_call_id": "call_1",
+ "name": "search_articles",
+ },
+ {
+ "role": "assistant",
+ "content": "I found 20 troubleshooting articles, and there are more available. "
+ "The first one is 'How to troubleshoot login issues'. "
+ "Would you like to see more results?",
+ },
+ ],
+ )
+
+ return suite
diff --git a/toolkits/zendesk/evals/eval_tickets.py b/toolkits/zendesk/evals/eval_tickets.py
new file mode 100644
index 00000000..050407e0
--- /dev/null
+++ b/toolkits/zendesk/evals/eval_tickets.py
@@ -0,0 +1,631 @@
+from arcade_evals import (
+ BinaryCritic,
+ EvalRubric,
+ EvalSuite,
+ ExpectedToolCall,
+ SimilarityCritic,
+ tool_eval,
+)
+from arcade_tdk import ToolCatalog
+
+import arcade_zendesk
+from arcade_zendesk.enums import SortOrder, TicketStatus
+from arcade_zendesk.tools.tickets import (
+ add_ticket_comment,
+ get_ticket_comments,
+ list_tickets,
+ mark_ticket_solved,
+)
+
+# Evaluation rubric
+rubric = EvalRubric(
+ fail_threshold=0.85,
+ warn_threshold=0.95,
+)
+
+catalog = ToolCatalog()
+catalog.add_module(arcade_zendesk)
+
+
+@tool_eval()
+def zendesk_tickets_read_eval_suite() -> EvalSuite:
+ """Evaluation suite for ticket reading operations."""
+ suite = EvalSuite(
+ name="Zendesk Tickets Read Operations",
+ system_message=(
+ "You are an AI assistant with access to Zendesk ticket tools. "
+ "Use them to help users view and manage support tickets."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Basic ticket listing
+ suite.add_case(
+ name="List all open tickets",
+ user_message="Show me all open tickets",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={},
+ )
+ ],
+ rubric=rubric,
+ critics=[], # No args to validate
+ )
+
+ suite.add_case(
+ name="List tickets with explicit status request",
+ user_message="Can you list the open support tickets?",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={},
+ )
+ ],
+ rubric=rubric,
+ critics=[],
+ )
+
+ suite.add_case(
+ name="Request for ticket overview",
+ user_message="I need to see what tickets are currently open",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={},
+ )
+ ],
+ rubric=rubric,
+ critics=[],
+ )
+
+ # Test pagination
+ suite.add_case(
+ name="List tickets with limit",
+ user_message="Show me the first 5 open tickets",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={"limit": 5},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="limit", weight=1.0),
+ ],
+ )
+
+ # Test status filter
+ suite.add_case(
+ name="List tickets with specific status",
+ user_message="Show me all pending tickets",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={"status": TicketStatus.PENDING},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="status", weight=1.0),
+ ],
+ )
+
+ # Test sort order
+ suite.add_case(
+ name="List tickets oldest first",
+ user_message="Show me tickets sorted from oldest to newest",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=list_tickets,
+ args={"sort_order": SortOrder.ASC},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="sort_order", weight=1.0),
+ ],
+ )
+
+ return suite
+
+
+@tool_eval()
+def zendesk_get_ticket_comments_eval_suite() -> EvalSuite:
+ """Evaluation suite for getting ticket comments."""
+ suite = EvalSuite(
+ name="Zendesk Get Ticket Comments",
+ system_message=(
+ "You are an AI assistant with access to Zendesk ticket tools. "
+ "Use them to help users view ticket comments and conversation history."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Get comments for a ticket
+ suite.add_case(
+ name="Get comments for specific ticket",
+ user_message="Show me the comments for ticket 123",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=get_ticket_comments,
+ args={"ticket_id": 123},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ )
+
+ suite.add_case(
+ name="View ticket conversation",
+ user_message="Can you show me the conversation history for ticket #456?",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=get_ticket_comments,
+ args={"ticket_id": 456},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ )
+
+ suite.add_case(
+ name="Get ticket description",
+ user_message="What is the original description of ticket 789?",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=get_ticket_comments,
+ args={"ticket_id": 789},
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ )
+
+ return suite
+
+
+@tool_eval()
+def zendesk_ticket_comments_eval_suite() -> EvalSuite:
+ """Evaluation suite for ticket comment operations."""
+ suite = EvalSuite(
+ name="Zendesk Ticket Comments",
+ system_message=(
+ "You are an AI assistant with access to Zendesk ticket tools. "
+ "Use them to help users add comments to support tickets."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Public comments
+ suite.add_case(
+ name="Add public comment to ticket",
+ user_message="Add a comment to ticket 123 saying 'We are investigating this issue'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 123,
+ "comment_body": "We are investigating this issue",
+ "public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="public", weight=0.2),
+ ],
+ )
+
+ suite.add_case(
+ name="Add public comment without specifying visibility",
+ user_message="Please comment on ticket #456: "
+ "The issue has been escalated to our engineering team",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 456,
+ "comment_body": "The issue has been escalated to our engineering team",
+ "public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="public", weight=0.2),
+ ],
+ )
+
+ # Internal comments
+ suite.add_case(
+ name="Add internal comment to ticket",
+ user_message="Add an internal note to ticket 789: Customer is VIP, prioritize this issue",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 789,
+ "comment_body": "Customer is VIP, prioritize this issue",
+ "public": False,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="public", weight=0.2),
+ ],
+ )
+
+ suite.add_case(
+ name="Add private comment to ticket",
+ user_message="Add a private comment to ticket 321 for agents only: "
+ "Check with backend team about API limits",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 321,
+ "comment_body": "Check with backend team about API limits",
+ "public": False,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="public", weight=0.2),
+ ],
+ )
+
+ # Complex comment scenarios
+ suite.add_case(
+ name="Add detailed public update",
+ user_message="Update ticket 555 with: 'We've identified the root cause. "
+ "A fix will be deployed within 24 hours. We apologize for the inconvenience.'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 555,
+ "comment_body": "We've identified the root cause. "
+ "A fix will be deployed within 24 hours. We apologize for the inconvenience.",
+ "public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.6),
+ BinaryCritic(critic_field="public", weight=0.1),
+ ],
+ )
+
+ return suite
+
+
+@tool_eval()
+def zendesk_ticket_resolution_eval_suite() -> EvalSuite:
+ """Evaluation suite for ticket resolution operations."""
+ suite = EvalSuite(
+ name="Zendesk Ticket Resolution",
+ system_message=(
+ "You are an AI assistant with access to Zendesk ticket tools. "
+ "Use them to help users resolve support tickets."
+ "Consider that closing a ticket is the same as marking it as solved."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Simple resolution
+ suite.add_case(
+ name="Mark ticket as solved without comment",
+ user_message="Mark ticket 100 as solved",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 100,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ )
+
+ suite.add_case(
+ name="Close ticket",
+ user_message="Please close ticket #200",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 200,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ )
+
+ # Resolution with public comment
+ suite.add_case(
+ name="Solve ticket with public resolution comment",
+ user_message="Resolve ticket 300 with comment: "
+ "'Issue resolved by updating your account settings'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 300,
+ "comment_body": "Issue resolved by updating your account settings",
+ "comment_public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="comment_public", weight=0.2),
+ ],
+ )
+
+ suite.add_case(
+ name="Close ticket with customer-facing message",
+ user_message="Close ticket 400 and tell the customer: "
+ "Your refund has been processed successfully",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 400,
+ "comment_body": "Your refund has been processed successfully",
+ "comment_public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="comment_public", weight=0.2),
+ ],
+ )
+
+ # Resolution with internal comment
+ suite.add_case(
+ name="Solve ticket with internal note",
+ user_message="Mark ticket 500 as solved with internal note: "
+ "'Resolved via backend database fix'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 500,
+ "comment_body": "Resolved via backend database fix",
+ "comment_public": False,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="comment_public", weight=0.2),
+ ],
+ )
+
+ # Default internal comment behavior
+ suite.add_case(
+ name="Solve ticket with comment defaults to internal",
+ user_message="Mark ticket 550 as solved with comment: 'Fixed by applying patch #2345'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 550,
+ "comment_body": "Fixed by applying patch #2345",
+ # comment_public should default to False if not specified
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.4),
+ SimilarityCritic(critic_field="comment_body", weight=0.6),
+ ],
+ )
+
+ suite.add_case(
+ name="Close ticket with private resolution details",
+ user_message="Close ticket 600 with a private note for agents: "
+ "'Customer account had duplicate entries, merged successfully'",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 600,
+ "comment_body": "Customer account had duplicate entries, merged successfully",
+ "comment_public": False,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="comment_public", weight=0.2),
+ ],
+ )
+
+ return suite
+
+
+@tool_eval()
+def zendesk_ticket_workflow_eval_suite() -> EvalSuite:
+ """Evaluation suite for ticket workflow scenarios with context."""
+ suite = EvalSuite(
+ name="Zendesk Ticket Workflows",
+ system_message=(
+ "You are an AI assistant with access to Zendesk ticket tools. "
+ "Use them to help users manage support ticket workflows."
+ ),
+ catalog=catalog,
+ rubric=rubric,
+ )
+
+ # Workflow: View then comment
+ suite.add_case(
+ name="Comment on specific ticket after viewing",
+ user_message="Add a comment to the login issue ticket saying we're working on it",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=add_ticket_comment,
+ args={
+ "ticket_id": 1,
+ "comment_body": "We're currently working on resolving your login issue.",
+ "public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="public", weight=0.2),
+ ],
+ additional_messages=[
+ {
+ "role": "user",
+ "content": "Show me all open tickets",
+ },
+ {
+ "role": "assistant",
+ "content": "I'll list all open tickets for you.",
+ "tool_calls": [
+ {
+ "id": "call_1",
+ "type": "function",
+ "function": {
+ "name": "list_tickets",
+ "arguments": "{}",
+ },
+ }
+ ],
+ },
+ {
+ "role": "tool",
+ "content": '{"tickets": [{"id": 1, "subject": "Login issue", "status": "open", '
+ '"html_url": "https://example.zendesk.com/agent/tickets/1"}, '
+ '{"id": 2, "subject": "Password reset request", "status": "open", '
+ '"html_url": "https://example.zendesk.com/agent/tickets/2"}], "count": 2}',
+ "tool_call_id": "call_1",
+ "name": "list_tickets",
+ },
+ {
+ "role": "assistant",
+ "content": "I found 2 open tickets:\n"
+ "1. Ticket #1: Login issue\n2. Ticket #2: Password reset request",
+ },
+ ],
+ )
+
+ # Workflow: Comment then resolve
+ suite.add_case(
+ name="Resolve ticket after adding solution",
+ user_message="Now mark that ticket as solved",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 789,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=1.0),
+ ],
+ additional_messages=[
+ {
+ "role": "user",
+ "content": "Add a comment to ticket 789: "
+ "'Reset your password using the forgot password link on the login page'",
+ },
+ {
+ "role": "assistant",
+ "content": "I'll add that comment to ticket 789.",
+ "tool_calls": [
+ {
+ "id": "call_1",
+ "type": "function",
+ "function": {
+ "name": "add_ticket_comment",
+ "arguments": '{"ticket_id": 789, "comment_body": '
+ '"Reset your password using the forgot password link on the login '
+ 'page", "public": true}',
+ },
+ }
+ ],
+ },
+ {
+ "role": "tool",
+ "content": '{"success": true, "ticket_id": 789, "comment_type": "public", '
+ '"ticket": {"id": 789, "html_url": "https://example.zendesk.com/agent/tickets/789"}}',
+ "tool_call_id": "call_1",
+ "name": "add_ticket_comment",
+ },
+ {
+ "role": "assistant",
+ "content": "I've added the comment with password reset instructions "
+ "to ticket #789.",
+ },
+ ],
+ )
+
+ # Workflow: Multiple updates
+ suite.add_case(
+ name="Add final comment and close ticket",
+ user_message="Add 'This issue has been fully resolved' and close ticket 999",
+ expected_tool_calls=[
+ ExpectedToolCall(
+ func=mark_ticket_solved,
+ args={
+ "ticket_id": 999,
+ "comment_body": "This issue has been fully resolved",
+ "comment_public": True,
+ },
+ )
+ ],
+ rubric=rubric,
+ critics=[
+ BinaryCritic(critic_field="ticket_id", weight=0.3),
+ SimilarityCritic(critic_field="comment_body", weight=0.5),
+ BinaryCritic(critic_field="comment_public", weight=0.2),
+ ],
+ )
+
+ return suite
diff --git a/toolkits/zendesk/pyproject.toml b/toolkits/zendesk/pyproject.toml
new file mode 100644
index 00000000..e5162bde
--- /dev/null
+++ b/toolkits/zendesk/pyproject.toml
@@ -0,0 +1,59 @@
+[build-system]
+requires = [ "hatchling",]
+build-backend = "hatchling.build"
+
+[project]
+name = "arcade_zendesk"
+version = "0.1.0"
+requires-python = ">=3.10"
+dependencies = [
+ "arcade-tdk>=2.0.0,<3.0.0",
+ "httpx>=0.25.0,<1.0.0",
+ "beautifulsoup4>=4.0.0,<5"
+]
+
+
+[project.optional-dependencies]
+dev = [
+ "arcade-ai[evals]>=2.0.0,<3.0.0",
+ "arcade-serve>=2.0.0,<3.0.0",
+ "pytest>=8.3.0,<8.4.0",
+ "pytest-cov>=4.0.0,<4.1.0",
+ "pytest-mock>=3.11.1,<3.12.0",
+ "pytest-asyncio>=0.24.0,<0.25.0",
+ "mypy>=1.5.1,<1.6.0",
+ "pre-commit>=3.4.0,<3.5.0",
+ "tox>=4.11.1,<4.12.0",
+ "ruff>=0.7.4,<0.8.0",
+]
+
+# Use local path sources for arcade libs when working locally
+[tool.uv.sources]
+arcade-ai = { path = "../../", editable = true }
+arcade-serve = { path = "../../libs/arcade-serve/", editable = true }
+arcade-tdk = { path = "../../libs/arcade-tdk/", editable = true }
+
+
+[tool.mypy]
+files = [ "arcade_zendesk/**/*.py",]
+python_version = "3.10"
+disallow_untyped_defs = "True"
+disallow_any_unimported = "True"
+no_implicit_optional = "True"
+check_untyped_defs = "True"
+warn_return_any = "True"
+warn_unused_ignores = "True"
+show_error_codes = "True"
+ignore_missing_imports = "True"
+
+[tool.pytest.ini_options]
+testpaths = [ "tests",]
+
+[tool.coverage.report]
+skip_empty = true
+
+[tool.ruff.lint]
+ignore = ["C901"]
+
+[tool.hatch.build.targets.wheel]
+packages = [ "arcade_zendesk",]
diff --git a/toolkits/zendesk/tests/__init__.py b/toolkits/zendesk/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/toolkits/zendesk/tests/conftest.py b/toolkits/zendesk/tests/conftest.py
new file mode 100644
index 00000000..ad98f5cf
--- /dev/null
+++ b/toolkits/zendesk/tests/conftest.py
@@ -0,0 +1,84 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from arcade_tdk import ToolContext
+
+
+@pytest.fixture
+def mock_context():
+ """Standard mock context fixture used across all arcade toolkits."""
+ context = MagicMock(spec=ToolContext)
+
+ context.get_auth_token_or_empty = MagicMock(return_value="fake-token")
+ context.get_secret = MagicMock()
+
+ return context
+
+
+@pytest.fixture
+def mock_httpx_client(mocker):
+ """Mock httpx.AsyncClient for API calls."""
+ mock_client_class = mocker.patch("httpx.AsyncClient", autospec=True)
+ mock_client = AsyncMock()
+ mock_client_class.return_value.__aenter__.return_value = mock_client
+ return mock_client
+
+
+@pytest.fixture
+def sample_article_response():
+ """Sample article data for testing."""
+ return {
+ "id": 123456,
+ "title": "How to reset your password",
+ "body": "To reset your password, follow these steps:
"
+ "- Click forgot password
- Enter your email
",
+ "url": "https://support.example.com/hc/en-us/articles/123456",
+ "created_at": "2024-01-15T10:00:00Z",
+ "updated_at": "2024-06-01T15:30:00Z",
+ "section_id": 789,
+ "category_id": 456,
+ "label_names": ["password", "security", "account"],
+ }
+
+
+@pytest.fixture
+def build_search_response(sample_article_response):
+ """Builder for search API responses."""
+
+ def builder(articles=None, next_page=None, count=None):
+ if articles is None:
+ articles = [sample_article_response]
+
+ response = {
+ "results": articles,
+ "next_page": next_page,
+ "page": 1,
+ "per_page": len(articles),
+ "page_count": 1,
+ }
+
+ if count is not None:
+ response["count"] = count
+
+ return response
+
+ return builder
+
+
+@pytest.fixture
+def mock_http_response():
+ """Factory for creating mock HTTP responses."""
+
+ def create_response(json_data=None, status_code=200, raise_for_status=True):
+ response = MagicMock()
+ response.json.return_value = json_data
+ response.status_code = status_code
+
+ if raise_for_status and status_code >= 400:
+ response.raise_for_status.side_effect = Exception(f"HTTP {status_code}")
+ else:
+ response.raise_for_status.return_value = None
+
+ return response
+
+ return create_response
diff --git a/toolkits/zendesk/tests/test_search_articles.py b/toolkits/zendesk/tests/test_search_articles.py
new file mode 100644
index 00000000..bb357cd2
--- /dev/null
+++ b/toolkits/zendesk/tests/test_search_articles.py
@@ -0,0 +1,423 @@
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from arcade_tdk.errors import RetryableToolError, ToolExecutionError
+
+from arcade_zendesk.enums import ArticleSortBy, SortOrder
+from arcade_zendesk.tools.search_articles import search_articles
+
+
+class TestSearchArticlesValidation:
+ """Test input validation for search_articles."""
+
+ @pytest.mark.asyncio
+ async def test_missing_subdomain(self, mock_context):
+ """Test error when subdomain is not configured."""
+ mock_context.get_secret.side_effect = ValueError("Secret not found")
+
+ with pytest.raises(ToolExecutionError) as exc_info:
+ await search_articles(context=mock_context, query="test")
+
+ assert "subdomain is not set" in str(exc_info.value.message)
+
+ @pytest.mark.asyncio
+ async def test_missing_search_params(self, mock_context):
+ """Test error when no search parameters provided."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ with pytest.raises(RetryableToolError) as exc_info:
+ await search_articles(context=mock_context)
+
+ assert "At least one search parameter" in str(exc_info.value.message)
+
+ @pytest.mark.parametrize(
+ "date_param,date_value",
+ [
+ ("created_after", "2024/01/01"),
+ ("created_before", "01-15-2024"),
+ ("created_at", "2024-1-15"),
+ ("created_after", "2024-01-1"),
+ ("created_before", "20240115"),
+ ("created_at", "not-a-date"),
+ ],
+ )
+ @pytest.mark.asyncio
+ async def test_invalid_date_format(self, mock_context, date_param, date_value):
+ """Test validation of date format parameters."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ with pytest.raises(RetryableToolError) as exc_info:
+ await search_articles(context=mock_context, query="test", **{date_param: date_value})
+
+ assert "Invalid date format" in str(exc_info.value.message)
+ assert "YYYY-MM-DD" in str(exc_info.value.message)
+ assert date_param in str(exc_info.value.message)
+
+ @pytest.mark.parametrize("limit", [0, -1, -10])
+ @pytest.mark.asyncio
+ async def test_invalid_limit(self, mock_context, limit):
+ """Test validation of limit parameter."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ with pytest.raises(RetryableToolError) as exc_info:
+ await search_articles(context=mock_context, query="test", limit=limit)
+
+ assert "at least 1" in str(exc_info.value.message)
+
+ @pytest.mark.parametrize("offset", [-1, -10])
+ @pytest.mark.asyncio
+ async def test_invalid_offset(self, mock_context, offset):
+ """Test validation of offset parameter."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ with pytest.raises(RetryableToolError) as exc_info:
+ await search_articles(context=mock_context, query="test", offset=offset)
+
+ assert "cannot be negative" in str(exc_info.value.message)
+
+
+class TestSearchArticlesSuccess:
+ """Test successful search scenarios."""
+
+ @pytest.mark.asyncio
+ async def test_basic_search(
+ self, mock_context, mock_httpx_client, build_search_response, mock_http_response
+ ):
+ """Test basic search with query parameter."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup mock response
+ search_response = build_search_response()
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(context=mock_context, query="password reset")
+
+ assert "results" in result
+ assert len(result["results"]) == 1
+ assert result["results"][0]["metadata"]["title"] == "How to reset your password"
+
+ mock_httpx_client.get.assert_called_once()
+ call_args = mock_httpx_client.get.call_args
+ assert (
+ call_args[0][0]
+ == "https://test-subdomain.zendesk.com/api/v2/help_center/articles/search"
+ )
+ assert call_args[1]["params"]["query"] == "password reset"
+ # Check that pagination params were set correctly
+ assert call_args[1]["params"]["page"] == 1
+ assert call_args[1]["params"]["per_page"] == 100
+
+ @pytest.mark.asyncio
+ async def test_search_with_filters(
+ self, mock_context, mock_httpx_client, build_search_response, mock_http_response
+ ):
+ """Test search with multiple filter parameters."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ search_response = build_search_response()
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(
+ context=mock_context,
+ query="API",
+ created_after="2024-01-01",
+ sort_by=ArticleSortBy.CREATED_AT,
+ sort_order=SortOrder.DESC,
+ limit=25,
+ )
+
+ assert "results" in result
+
+ # Verify all parameters were passed
+ call_params = mock_httpx_client.get.call_args[1]["params"]
+ assert call_params["query"] == "API"
+ assert call_params["created_after"] == "2024-01-01"
+ assert call_params["sort_by"] == "created_at"
+ assert call_params["sort_order"] == "desc"
+ # Should fetch first page with 100 items per page
+ assert call_params["page"] == 1
+ assert call_params["per_page"] == 100
+
+ @pytest.mark.asyncio
+ async def test_search_without_body(
+ self,
+ mock_context,
+ mock_httpx_client,
+ sample_article_response,
+ mock_http_response,
+ ):
+ """Test search with include_body=False."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ search_response = {"results": [sample_article_response], "next_page": None}
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(context=mock_context, query="test", include_body=False)
+
+ assert result["results"][0]["content"] is None
+ assert result["results"][0]["metadata"]["title"] == sample_article_response["title"]
+
+ @pytest.mark.asyncio
+ async def test_search_by_labels(
+ self, mock_context, mock_httpx_client, build_search_response, mock_http_response
+ ):
+ """Test search by label names."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ search_response = build_search_response()
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(context=mock_context, label_names=["password", "security"])
+
+ assert "results" in result
+ assert mock_httpx_client.get.call_args[1]["params"]["label_names"] == "password,security"
+
+
+class TestSearchArticlesPagination:
+ """Test pagination scenarios."""
+
+ @pytest.mark.asyncio
+ async def test_single_page_default(
+ self, mock_context, mock_httpx_client, build_search_response, mock_http_response
+ ):
+ """Test default behavior returns single page."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ search_response = build_search_response(count=100)
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(context=mock_context, query="test")
+
+ assert len(result["results"]) == 1
+ assert mock_httpx_client.get.call_count == 1
+
+ @pytest.mark.asyncio
+ async def test_fetch_with_limit_across_pages(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test fetching results across multiple pages with limit."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup pagination responses - 100 items per page
+ articles_page1 = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(1, 101)
+ ]
+ articles_page2 = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(101, 201)
+ ]
+
+ page1 = {"results": articles_page1, "next_page": "page2"}
+ page2 = {"results": articles_page2, "next_page": "page3"}
+
+ mock_httpx_client.get.side_effect = [
+ mock_http_response(page1),
+ mock_http_response(page2),
+ ]
+
+ # Request 150 items starting from offset 0
+ result = await search_articles(context=mock_context, query="test", limit=150)
+
+ assert result["count"] == 150
+ assert "next_offset" in result # More results available
+ assert result["next_offset"] == 150
+ assert mock_httpx_client.get.call_count == 2 # Fetched 2 pages
+
+ @pytest.mark.asyncio
+ async def test_fetch_with_offset(self, mock_context, mock_httpx_client, mock_http_response):
+ """Test fetching with offset parameter."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup response - page 2 would have items 101-200
+ # We want items starting from offset 150 (which is item 151, at index 50 on page 2)
+ articles_page2 = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(101, 201)
+ ]
+ response = {"results": articles_page2, "next_page": "page3"}
+
+ mock_httpx_client.get.return_value = mock_http_response(response)
+
+ # Request 30 items starting from offset 150
+ result = await search_articles(context=mock_context, query="test", offset=150, limit=30)
+
+ assert result["count"] == 30
+ assert "next_offset" in result
+ assert result["next_offset"] == 180
+
+ # Should request page 2 (offset 150 = page 2, starting at index 50)
+ call_params = mock_httpx_client.get.call_args[1]["params"]
+ assert call_params["page"] == 2
+
+ @pytest.mark.asyncio
+ async def test_no_next_offset_when_no_more_results(
+ self, mock_context, mock_httpx_client, build_search_response, mock_http_response
+ ):
+ """Test that next_offset is not included when no more results."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup response with no next page
+ articles = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(1, 21)
+ ]
+ response = {"results": articles, "next_page": None}
+
+ mock_httpx_client.get.return_value = mock_http_response(response)
+
+ result = await search_articles(context=mock_context, query="test", limit=20)
+
+ assert result["count"] == 20
+ assert "next_offset" not in result # No more results
+
+ @pytest.mark.asyncio
+ async def test_partial_page_with_more_items(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test that next_offset is included when there are more items on the current page."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup response with 50 items on a page, but we only request 30
+ articles = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(1, 51)
+ ]
+ response = {"results": articles, "next_page": None}
+
+ mock_httpx_client.get.return_value = mock_http_response(response)
+
+ # Request only 30 items when page has 50
+ result = await search_articles(context=mock_context, query="test", limit=30)
+
+ assert result["count"] == 30
+ assert "next_offset" in result # More items available on current page
+ assert result["next_offset"] == 30
+
+ @pytest.mark.asyncio
+ async def test_request_more_than_available(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test when requesting more items than are available returns only what's available."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Setup response with only 15 items total
+ articles = [
+ {"id": i, "title": f"Article {i}", "body": f"Content {i}"} for i in range(1, 16)
+ ]
+ response = {"results": articles, "next_page": None}
+
+ mock_httpx_client.get.return_value = mock_http_response(response)
+
+ # Request 30 items when only 15 are available
+ result = await search_articles(context=mock_context, query="test", limit=30)
+
+ assert result["count"] == 15 # Only returns what's available
+ assert "next_offset" not in result # No more results
+
+
+class TestSearchArticlesErrors:
+ """Test error handling scenarios."""
+
+ @pytest.mark.parametrize(
+ "status_code,error_key",
+ [
+ (400, "HTTP 400"),
+ (401, "HTTP 401"),
+ (403, "HTTP 403"),
+ (404, "HTTP 404"),
+ (500, "HTTP 500"),
+ ],
+ )
+ @pytest.mark.asyncio
+ async def test_http_errors(self, mock_context, mock_httpx_client, status_code, error_key):
+ """Test handling of HTTP errors."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Create mock error response
+ error_response = MagicMock()
+ error_response.status_code = status_code
+ error_response.text = f"Error message for {status_code}"
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ message=f"HTTP {status_code}", request=MagicMock(), response=error_response
+ )
+
+ mock_httpx_client.get.return_value = error_response
+
+ with pytest.raises(ToolExecutionError) as exc_info:
+ await search_articles(context=mock_context, query="test")
+
+ assert "Failed to search articles" in str(exc_info.value.message)
+ assert f"HTTP {status_code}" in str(exc_info.value.message)
+
+ @pytest.mark.asyncio
+ async def test_timeout_error(self, mock_context, mock_httpx_client):
+ """Test handling of timeout errors."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.get.side_effect = httpx.TimeoutException("Request timed out")
+
+ with pytest.raises(RetryableToolError) as exc_info:
+ await search_articles(context=mock_context, query="test")
+
+ assert "timed out" in str(exc_info.value.message)
+ assert exc_info.value.retry_after_ms == 5000
+
+ @pytest.mark.asyncio
+ async def test_unexpected_error(self, mock_context, mock_httpx_client):
+ """Test handling of unexpected errors."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.get.side_effect = Exception("Unexpected error occurred")
+
+ with pytest.raises(ToolExecutionError) as exc_info:
+ await search_articles(context=mock_context, query="test")
+
+ assert "Unexpected error occurred" in str(exc_info.value.message)
+
+
+class TestSearchArticlesContentProcessing:
+ """Test content processing and formatting."""
+
+ @pytest.mark.asyncio
+ async def test_html_cleaning(self, mock_context, mock_httpx_client, mock_http_response):
+ """Test HTML content is properly cleaned."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ article_with_html = {
+ "id": 1,
+ "title": "Test Article",
+ "body": "Header
Paragraph with bold and "
+ "italic.
Div content
",
+ "url": "https://example.com/article/1",
+ }
+
+ search_response = {"results": [article_with_html], "next_page": None}
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ result = await search_articles(context=mock_context, query="test", include_body=True)
+
+ content = result["results"][0]["content"]
+ assert content == "Header Paragraph with bold and italic . Div content"
+
+ @pytest.mark.asyncio
+ async def test_max_article_length(self, mock_context, mock_httpx_client, mock_http_response):
+ """Test article length limiting."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ long_article = {
+ "id": 1,
+ "title": "Long Article",
+ "body": "A" * 1000, # 1000 character body
+ }
+
+ search_response = {"results": [long_article], "next_page": None}
+ mock_httpx_client.get.return_value = mock_http_response(search_response)
+
+ # Test with default 500 char limit
+ result = await search_articles(context=mock_context, query="test")
+ assert len(result["results"][0]["content"]) < 520 # 500 + truncation suffix
+
+ # Test with custom limit
+ result = await search_articles(context=mock_context, query="test", max_article_length=100)
+ assert len(result["results"][0]["content"]) < 120 # 100 + truncation suffix
+
+ # Test with no limit
+ result = await search_articles(context=mock_context, query="test", max_article_length=None)
+ assert len(result["results"][0]["content"]) == 1000
diff --git a/toolkits/zendesk/tests/test_tickets.py b/toolkits/zendesk/tests/test_tickets.py
new file mode 100644
index 00000000..1f10174d
--- /dev/null
+++ b/toolkits/zendesk/tests/test_tickets.py
@@ -0,0 +1,526 @@
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from arcade_core.errors import ToolExecutionError
+
+from arcade_zendesk.enums import SortOrder, TicketStatus
+from arcade_zendesk.tools.tickets import (
+ add_ticket_comment,
+ get_ticket_comments,
+ list_tickets,
+ mark_ticket_solved,
+)
+
+
+class TestListTickets:
+ """Test list_tickets functionality."""
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_success(self, mock_context, mock_httpx_client, mock_http_response):
+ """Test successful listing of open tickets."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Mock response data - includes url field that should be removed
+ tickets_response = {
+ "tickets": [
+ {
+ "id": 1,
+ "subject": "Login issue",
+ "status": "open",
+ "url": "https://test-subdomain.zendesk.com/api/v2/tickets/1.json",
+ },
+ {
+ "id": 2,
+ "subject": "Password reset request",
+ "status": "open",
+ "url": "https://test-subdomain.zendesk.com/api/v2/tickets/2.json",
+ },
+ ]
+ }
+
+ mock_httpx_client.get.return_value = mock_http_response(tickets_response)
+
+ result = await list_tickets(mock_context, status=TicketStatus.OPEN)
+
+ # Verify the result is structured data
+ assert isinstance(result, dict)
+ assert "tickets" in result
+ assert "count" in result
+ assert result["count"] == 2
+
+ # Verify tickets have html_url but not url
+ for ticket in result["tickets"]:
+ assert "url" not in ticket
+ assert "html_url" in ticket
+ assert ticket["html_url"].startswith(
+ "https://test-subdomain.zendesk.com/agent/tickets/"
+ )
+
+ # Verify the API call with default parameters
+ mock_httpx_client.get.assert_called()
+ # The fetch_paginated_results makes the actual call
+ call_args = mock_httpx_client.get.call_args
+ assert "https://test-subdomain.zendesk.com/api/v2/tickets.json" in call_args[0][0]
+ assert call_args[1]["params"]["status"] == "open"
+ assert call_args[1]["params"]["per_page"] == 100
+ assert call_args[1]["params"]["sort_order"] == "desc"
+ assert call_args[1]["params"]["page"] == 1 # First page
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_with_offset_limit(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test listing tickets with offset and limit."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Mock response for page 2 (offset 10, limit 5)
+ tickets_response = {
+ "tickets": [
+ {"id": 11, "subject": "Test 11", "status": "open"},
+ {"id": 12, "subject": "Test 12", "status": "open"},
+ {"id": 13, "subject": "Test 13", "status": "open"},
+ {"id": 14, "subject": "Test 14", "status": "open"},
+ {"id": 15, "subject": "Test 15", "status": "open"},
+ ],
+ "next_page": "https://test.zendesk.com/api/v2/tickets.json?page=3",
+ }
+
+ mock_httpx_client.get.return_value = mock_http_response(tickets_response)
+
+ result = await list_tickets(
+ mock_context, status=TicketStatus.OPEN, limit=5, offset=10, sort_order=SortOrder.ASC
+ )
+
+ # Verify response structure
+ assert result["count"] == 5
+ assert len(result["tickets"]) == 5
+ assert "next_offset" in result
+ assert result["next_offset"] == 15 # offset + limit
+
+ # Verify API call parameters
+ call_args = mock_httpx_client.get.call_args
+ assert (
+ call_args[1]["params"]["page"] == 2
+ ) # offset 10 / per_page 100 = page 2 (but adjusted for limit)
+ assert call_args[1]["params"]["per_page"] == 100
+ assert call_args[1]["params"]["sort_order"] == "asc"
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_no_more_results(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test listing tickets when no more results are available."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Mock response with no next_page - simulating the last page
+ tickets_response = {
+ "tickets": [{"id": 21, "subject": "Test", "status": "pending"}],
+ # No next_page means no more results
+ }
+
+ mock_httpx_client.get.return_value = mock_http_response(tickets_response)
+
+ result = await list_tickets(
+ mock_context,
+ status=TicketStatus.PENDING,
+ limit=10,
+ offset=0, # Start from beginning
+ )
+
+ # Verify no next_offset when no more results
+ assert "next_offset" not in result
+ assert result["count"] == 1
+
+ # Verify API call parameters
+ call_args = mock_httpx_client.get.call_args
+ assert call_args[1]["params"]["status"] == "pending"
+ assert call_args[1]["params"]["page"] == 1
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_no_tickets(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test when no tickets are found."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.get.return_value = mock_http_response({"tickets": []})
+
+ result = await list_tickets(mock_context, status=TicketStatus.OPEN)
+
+ assert result["tickets"] == []
+ assert result["count"] == 0
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_error(self, mock_context, mock_httpx_client):
+ """Test error handling for failed API call."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Mock error response that raise_for_status will catch
+ error_response = MagicMock()
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ "Unauthorized", request=MagicMock(), response=MagicMock(status_code=401)
+ )
+ mock_httpx_client.get.return_value = error_response
+
+ with pytest.raises(ToolExecutionError):
+ await list_tickets(mock_context)
+
+ @pytest.mark.asyncio
+ async def test_list_tickets_no_subdomain(self, mock_context):
+ """Test when subdomain is not configured."""
+ mock_context.get_secret.return_value = None
+
+ with pytest.raises(ToolExecutionError):
+ await list_tickets(mock_context)
+
+
+class TestGetTicketComments:
+ """Test get_ticket_comments functionality."""
+
+ @pytest.mark.asyncio
+ async def test_get_ticket_comments_success(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test successfully getting ticket comments."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # Mock response data
+ comments_response = {
+ "comments": [
+ {
+ "id": 1,
+ "body": "I cannot access my account. Please help!",
+ "author_id": 12345,
+ "created_at": "2024-01-15T10:00:00Z",
+ "public": True,
+ "attachments": [],
+ },
+ {
+ "id": 2,
+ "body": "I'll help you reset your password.",
+ "author_id": 67890,
+ "created_at": "2024-01-15T10:30:00Z",
+ "public": True,
+ "attachments": [
+ {
+ "file_name": "screenshot.png",
+ "content_url": "https://example.com/screenshot.png",
+ "size": 12345,
+ }
+ ],
+ },
+ ]
+ }
+
+ mock_httpx_client.get.return_value = mock_http_response(comments_response)
+
+ result = await get_ticket_comments(mock_context, ticket_id=123)
+
+ # Verify the result is structured data
+ assert isinstance(result, dict)
+ assert result["ticket_id"] == 123
+ assert result["count"] == 2
+ assert len(result["comments"]) == 2
+
+ # Verify attachments are included
+ assert result["comments"][1]["attachments"][0]["file_name"] == "screenshot.png"
+
+ # Verify the API call
+ mock_httpx_client.get.assert_called_once_with(
+ "https://test-subdomain.zendesk.com/api/v2/tickets/123/comments.json",
+ headers={
+ "Authorization": "Bearer fake-token",
+ "Content-Type": "application/json",
+ },
+ )
+
+ @pytest.mark.asyncio
+ async def test_get_ticket_comments_no_comments(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test when no comments are found."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.get.return_value = mock_http_response({"comments": []})
+
+ result = await get_ticket_comments(mock_context, ticket_id=123)
+
+ assert result["comments"] == []
+ assert result["count"] == 0
+
+ @pytest.mark.asyncio
+ async def test_get_ticket_comments_not_found(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test when ticket is not found."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.get.return_value = mock_http_response({}, status_code=404)
+
+ with pytest.raises(ToolExecutionError):
+ await get_ticket_comments(mock_context, ticket_id=999)
+
+ @pytest.mark.asyncio
+ async def test_get_ticket_comments_error(self, mock_context, mock_httpx_client):
+ """Test error handling when API fails."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ error_response = MagicMock()
+ error_response.status_code = 500
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ "Server Error", request=MagicMock(), response=MagicMock(status_code=500)
+ )
+
+ mock_httpx_client.get.return_value = error_response
+
+ with pytest.raises(ToolExecutionError):
+ await get_ticket_comments(mock_context, ticket_id=123)
+
+
+class TestAddTicketComment:
+ """Test add_ticket_comment functionality."""
+
+ @pytest.mark.asyncio
+ async def test_add_public_comment_success(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test successfully adding a public comment."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ ticket_response = {
+ "ticket": {
+ "id": 123,
+ "subject": "Test ticket",
+ "url": "https://test-subdomain.zendesk.com/api/v2/tickets/123.json",
+ }
+ }
+
+ mock_httpx_client.put.return_value = mock_http_response(ticket_response)
+
+ result = await add_ticket_comment(
+ mock_context,
+ ticket_id=123,
+ comment_body="This is a test comment",
+ public=True,
+ )
+
+ # Verify structured response
+ assert isinstance(result, dict)
+ assert result["success"] is True
+ assert result["ticket_id"] == 123
+ assert result["comment_type"] == "public"
+ assert "ticket" in result
+
+ # Verify ticket has html_url but not url
+ assert "url" not in result["ticket"]
+ assert "html_url" in result["ticket"]
+
+ # Verify the API call
+ mock_httpx_client.put.assert_called_once_with(
+ "https://test-subdomain.zendesk.com/api/v2/tickets/123.json",
+ headers={
+ "Authorization": "Bearer fake-token",
+ "Content-Type": "application/json",
+ },
+ json={"ticket": {"comment": {"body": "This is a test comment", "public": True}}},
+ )
+
+ @pytest.mark.asyncio
+ async def test_add_comment_default_public(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test that comment defaults to public when not specified."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.put.return_value = mock_http_response({"ticket": {"id": 123}})
+
+ result = await add_ticket_comment(
+ mock_context,
+ ticket_id=123,
+ comment_body="Test comment",
+ # Not specifying public parameter
+ )
+
+ assert result["comment_type"] == "public"
+
+ # Verify the API call has public=True
+ call_args = mock_httpx_client.put.call_args
+ assert call_args[1]["json"]["ticket"]["comment"]["public"] is True
+
+ @pytest.mark.asyncio
+ async def test_add_internal_comment_success(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test successfully adding an internal comment."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.put.return_value = mock_http_response({"ticket": {"id": 456}})
+
+ result = await add_ticket_comment(
+ mock_context,
+ ticket_id=456,
+ comment_body="Internal note for agents",
+ public=False,
+ )
+
+ assert result["comment_type"] == "internal"
+
+ @pytest.mark.asyncio
+ async def test_add_comment_error(self, mock_context, mock_httpx_client):
+ """Test error handling when adding comment fails."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ error_response = MagicMock()
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ "Not Found", request=MagicMock(), response=MagicMock(status_code=404)
+ )
+ mock_httpx_client.put.return_value = error_response
+
+ with pytest.raises(ToolExecutionError):
+ await add_ticket_comment(mock_context, ticket_id=999, comment_body="Test comment")
+
+
+class TestMarkTicketSolved:
+ """Test mark_ticket_solved functionality."""
+
+ @pytest.mark.asyncio
+ async def test_mark_solved_without_comment(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test marking ticket as solved without a comment."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ ticket_response = {
+ "ticket": {
+ "id": 789,
+ "status": "solved",
+ "url": "https://test-subdomain.zendesk.com/api/v2/tickets/789.json",
+ }
+ }
+
+ mock_httpx_client.put.return_value = mock_http_response(ticket_response)
+
+ result = await mark_ticket_solved(mock_context, ticket_id=789)
+
+ # Verify structured response
+ assert isinstance(result, dict)
+ assert result["success"] is True
+ assert result["ticket_id"] == 789
+ assert result["status"] == "solved"
+ assert "comment_added" not in result
+
+ # Verify ticket has html_url
+ assert "html_url" in result["ticket"]
+ assert "url" not in result["ticket"]
+
+ # Verify the API call
+ mock_httpx_client.put.assert_called_once_with(
+ "https://test-subdomain.zendesk.com/api/v2/tickets/789.json",
+ headers={
+ "Authorization": "Bearer fake-token",
+ "Content-Type": "application/json",
+ },
+ json={"ticket": {"status": "solved"}},
+ )
+
+ @pytest.mark.asyncio
+ async def test_mark_solved_with_public_comment(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test marking ticket as solved with a public comment."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.put.return_value = mock_http_response({"ticket": {"id": 123}})
+
+ result = await mark_ticket_solved(
+ mock_context,
+ ticket_id=123,
+ comment_body="Issue resolved by resetting password",
+ comment_public=True,
+ )
+
+ assert result["comment_added"] is True
+ assert result["comment_type"] == "public"
+
+ # Verify the request body includes the comment
+ call_args = mock_httpx_client.put.call_args
+ request_body = call_args[1]["json"]
+ assert request_body["ticket"]["status"] == "solved"
+ assert request_body["ticket"]["comment"]["body"] == "Issue resolved by resetting password"
+ assert request_body["ticket"]["comment"]["public"] is True
+
+ @pytest.mark.asyncio
+ async def test_mark_solved_with_comment_default_internal(
+ self, mock_context, mock_httpx_client, mock_http_response
+ ):
+ """Test marking ticket as solved with comment defaults to internal."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ mock_httpx_client.put.return_value = mock_http_response({"ticket": {"id": 555}})
+
+ result = await mark_ticket_solved(
+ mock_context,
+ ticket_id=555,
+ comment_body="Internal resolution note",
+ # Not specifying comment_public, should default to False
+ )
+
+ assert result["comment_added"] is True
+ assert result["comment_type"] == "internal"
+
+ # Verify the comment is internal by default
+ call_args = mock_httpx_client.put.call_args
+ request_body = call_args[1]["json"]
+ assert request_body["ticket"]["comment"]["public"] is False
+
+ @pytest.mark.asyncio
+ async def test_mark_solved_error(self, mock_context, mock_httpx_client):
+ """Test error handling when marking ticket as solved fails."""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ error_response = MagicMock()
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ "Forbidden", request=MagicMock(), response=MagicMock(status_code=403)
+ )
+ mock_httpx_client.put.return_value = error_response
+
+ with pytest.raises(ToolExecutionError):
+ await mark_ticket_solved(mock_context, ticket_id=999)
+
+
+class TestAuthenticationAndSecrets:
+ """Test authentication and secrets handling."""
+
+ @pytest.mark.asyncio
+ async def test_no_auth_token(self, mock_context, mock_httpx_client):
+ """Test behavior when auth token is empty."""
+ mock_context.get_auth_token_or_empty.return_value = ""
+ mock_context.get_secret.return_value = "test-subdomain"
+
+ # The tools should still attempt the API call with empty token
+ error_response = MagicMock()
+ error_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+ "Unauthorized", request=MagicMock(), response=MagicMock(status_code=401)
+ )
+ mock_httpx_client.get.return_value = error_response
+
+ with pytest.raises(ToolExecutionError):
+ await list_tickets(mock_context)
+
+ # Should be called with empty Bearer token
+ call_args = mock_httpx_client.get.call_args
+ assert call_args[1]["headers"]["Authorization"] == "Bearer "
+
+ @pytest.mark.asyncio
+ async def test_subdomain_from_secret(self, mock_context, mock_httpx_client, mock_http_response):
+ """Test that subdomain is correctly retrieved from secrets."""
+ mock_context.get_secret.return_value = "my-company"
+
+ mock_httpx_client.get.return_value = mock_http_response({"tickets": []})
+
+ await list_tickets(mock_context, status=TicketStatus.OPEN)
+
+ # Verify the correct subdomain was used
+ call_args = mock_httpx_client.get.call_args
+ assert "https://my-company.zendesk.com" in call_args[0][0]
diff --git a/toolkits/zendesk/tests/test_utils.py b/toolkits/zendesk/tests/test_utils.py
new file mode 100644
index 00000000..995fa52c
--- /dev/null
+++ b/toolkits/zendesk/tests/test_utils.py
@@ -0,0 +1,291 @@
+import pytest
+
+from arcade_zendesk.utils import (
+ clean_html_text,
+ process_article_body,
+ process_search_results,
+ truncate_text,
+ validate_date_format,
+)
+
+
+class TestCleanHtmlText:
+ """Test HTML cleaning functionality."""
+
+ def test_clean_simple_html(self):
+ """Test cleaning basic HTML tags."""
+ html = "Hello World
"
+ assert clean_html_text(html) == "Hello World"
+
+ def test_clean_complex_html(self):
+ """Test cleaning complex HTML with multiple tags."""
+ html = """
+ Title
+ Paragraph with emphasis and bold.
+
+
+ """
+ cleaned = clean_html_text(html)
+ assert "Title" in cleaned
+ assert "Paragraph with emphasis and bold" in cleaned
+ assert "Item 1" in cleaned
+ assert "Item 2" in cleaned
+ assert "Footer content" in cleaned
+ assert "" not in cleaned
+ assert "
" not in cleaned
+
+ def test_clean_html_with_special_chars(self):
+ """Test cleaning HTML with special characters."""
+ html = "Price: £100 & €120
"
+ cleaned = clean_html_text(html)
+ assert "£100" in cleaned or "100" in cleaned # Depends on BeautifulSoup version
+ assert "&" in cleaned
+ assert "€120" in cleaned or "120" in cleaned
+
+ @pytest.mark.parametrize(
+ "input_value,expected",
+ [
+ (None, ""),
+ ("", ""),
+ (" ", ""),
+ ("", ""),
+ ("
", ""),
+ ],
+ )
+ def test_clean_html_edge_cases(self, input_value, expected):
+ """Test edge cases for HTML cleaning."""
+ assert clean_html_text(input_value) == expected
+
+ def test_clean_html_preserves_line_breaks(self):
+ """Test that meaningful line breaks are preserved."""
+ html = "Line 1
Line 2
"
+ cleaned = clean_html_text(html)
+ # Should have text from both lines
+ assert "Line 1" in cleaned
+ assert "Line 2" in cleaned
+
+
+class TestTruncateText:
+ """Test text truncation functionality."""
+
+ def test_truncate_long_text(self):
+ """Test truncating text longer than max length."""
+ text = "This is a very long text that needs to be truncated"
+ result = truncate_text(text, 20)
+ assert result == "This ... [truncated]"
+ assert len(result) == 20
+
+ def test_no_truncation_needed(self):
+ """Test text shorter than max length is not truncated."""
+ text = "Short text"
+ assert truncate_text(text, 20) == text
+
+ def test_truncate_with_custom_suffix(self):
+ """Test truncation with custom suffix."""
+ text = "This is a long text for testing"
+ result = truncate_text(text, 15, "...")
+ assert result == "This is a lo..."
+ assert len(result) == 15
+
+ @pytest.mark.parametrize(
+ "text,max_length,expected",
+ [
+ (None, 10, None),
+ ("", 10, ""),
+ ("Hello", 5, "Hello"),
+ ("Hello World", 5, " ... [truncated]"), # Suffix is longer than allowed
+ ],
+ )
+ def test_truncate_edge_cases(self, text, max_length, expected):
+ """Test edge cases for truncation."""
+ result = truncate_text(text, max_length)
+ if expected == " ... [truncated]":
+ # When suffix is longer than max_length, only suffix is returned
+ assert result == expected
+ else:
+ assert result == expected
+
+ def test_truncate_at_word_boundary(self):
+ """Test that truncation happens cleanly."""
+ text = "The quick brown fox jumps over the lazy dog"
+ result = truncate_text(text, 25)
+ assert result == "The quick ... [truncated]"
+ assert len(result) == 25
+
+
+class TestProcessArticleBody:
+ """Test article body processing."""
+
+ def test_process_html_body(self):
+ """Test processing HTML body content."""
+ body = "Article Title
Article content with formatting.
"
+ result = process_article_body(body)
+ assert "Article Title" in result
+ assert "Article content with formatting" in result
+ assert "" not in result
+ assert "" not in result
+
+ def test_process_body_with_truncation(self):
+ """Test processing body with max length."""
+ body = "" + "Long content " * 50 + "
"
+ result = process_article_body(body, max_length=100)
+ assert len(result) <= 100 + len(" ... [truncated]")
+ assert result.endswith(" ... [truncated]")
+
+ @pytest.mark.parametrize(
+ "body,max_length,expected",
+ [
+ (None, None, None),
+ ("", None, None),
+ ("Short
", 100, "Short"),
+ (
+ "",
+ None,
+ "",
+ ), # Empty paragraph returns empty string after cleaning
+ ],
+ )
+ def test_process_body_edge_cases(self, body, max_length, expected):
+ """Test edge cases for body processing."""
+ result = process_article_body(body, max_length)
+ assert result == expected
+
+
+class TestProcessSearchResults:
+ """Test search results processing."""
+
+ def test_process_results_with_body(self):
+ """Test processing results with body content included."""
+ results = [
+ {
+ "id": 1,
+ "title": "Article 1",
+ "body": "Content 1
",
+ "url": "https://example.com/1",
+ },
+ {
+ "id": 2,
+ "title": "Article 2",
+ "body": "Content 2
",
+ "url": "https://example.com/2",
+ },
+ ]
+
+ processed = process_search_results(results, include_body=True)
+
+ assert len(processed) == 2
+ assert processed[0]["content"] == "Content 1"
+ assert processed[0]["metadata"]["id"] == 1
+ assert processed[0]["metadata"]["title"] == "Article 1"
+ assert "body" not in processed[0]["metadata"]
+
+ assert processed[1]["content"] == "Content 2"
+ assert processed[1]["metadata"]["id"] == 2
+
+ def test_process_results_without_body(self):
+ """Test processing results without body content."""
+ results = [
+ {
+ "id": 1,
+ "title": "Article 1",
+ "body": "Content 1
",
+ "url": "https://example.com/1",
+ }
+ ]
+
+ processed = process_search_results(results, include_body=False)
+
+ assert processed[0]["content"] is None
+ assert processed[0]["metadata"]["id"] == 1
+ assert processed[0]["metadata"]["title"] == "Article 1"
+ assert "body" not in processed[0]["metadata"]
+
+ def test_process_results_with_max_body_length(self):
+ """Test processing results with body length limit."""
+ results = [
+ {
+ "id": 1,
+ "title": "Article",
+ "body": "" + "Long content " * 100 + "
",
+ }
+ ]
+
+ processed = process_search_results(results, include_body=True, max_body_length=50)
+
+ content = processed[0]["content"]
+ assert len(content) <= 50 + len(" ... [truncated]")
+ assert content.endswith(" ... [truncated]")
+
+ def test_process_empty_results(self):
+ """Test processing empty results list."""
+ processed = process_search_results([])
+ assert processed == []
+
+ def test_process_results_preserves_all_metadata(self):
+ """Test that all non-body fields are preserved in metadata."""
+ results = [
+ {
+ "id": 1,
+ "title": "Article",
+ "body": "Content
",
+ "url": "https://example.com/1",
+ "created_at": "2024-01-01",
+ "custom_field": "value",
+ "nested": {"key": "value"},
+ }
+ ]
+
+ processed = process_search_results(results, include_body=True)
+
+ metadata = processed[0]["metadata"]
+ assert metadata["id"] == 1
+ assert metadata["title"] == "Article"
+ assert metadata["url"] == "https://example.com/1"
+ assert metadata["created_at"] == "2024-01-01"
+ assert metadata["custom_field"] == "value"
+ assert metadata["nested"] == {"key": "value"}
+ assert "body" not in metadata
+
+
+class TestValidateDateFormat:
+ """Test date format validation."""
+
+ @pytest.mark.parametrize(
+ "date_string",
+ [
+ "2024-01-15",
+ "2024-12-31",
+ "2000-01-01",
+ "1999-12-31",
+ "2030-06-15",
+ ],
+ )
+ def test_valid_date_formats(self, date_string):
+ """Test valid YYYY-MM-DD date formats."""
+ assert validate_date_format(date_string) is True
+
+ @pytest.mark.parametrize(
+ "date_string",
+ [
+ "2024/01/15",
+ "01-15-2024",
+ "2024-1-15",
+ "2024-01-1",
+ "24-01-15",
+ "2024.01.15",
+ "20240115",
+ "January 15, 2024",
+ "15/01/2024",
+ "2024",
+ "2024-01",
+ "",
+ "not-a-date",
+ # Note: These have valid format but invalid values - regex only checks format
+ ],
+ )
+ def test_invalid_date_formats(self, date_string):
+ """Test invalid date formats."""
+ assert validate_date_format(date_string) is False