admin-valentin/arcade-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Zendesk Toolkit (#458)

Browse source 
This PR adds a new toolkit for integrating with Zendesk Help Center and
ticketing system.

  Features

- Search Articles: Search Help Center knowledge base with filters for
text, categories, labels, and dates
  - List Tickets: Retrieve all open support tickets
- Get Comments: Retrieve all comments for a specific ticket to
understand the context
  - Add Comments: Add public or internal comments to existing tickets
- Solve Tickets: Mark tickets as solved with optional internal
resolution comments

  Testing & Quality
  - Comprehensive test suite 
  - Evaluation scripts for real-world testing
  - All tests passing (make test)
  - Code quality checks passing (make check)
  - All evals passing

---------

Co-authored-by: “lgesuellip” <“lgesuellipinto@uade.edu.ar”>
Co-authored-by: lgesuellip <102637283+lgesuellip@users.noreply.github.com>
Co-authored-by: “lgesuellip” <lgesuellipinto@uade.edu.ar>
This commit is contained in:
Lucas Petralli 2025-07-17 17:34:26 -03:00 committed by GitHub
parent e1cf58fe1b
commit 1843e411f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 3407 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
toolkits/zendesk/.pre-commit-config.yaml Normal file
View file

@ -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

46
toolkits/zendesk/.ruff.toml Normal file
View file

@ -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

55
toolkits/zendesk/Makefile Normal file
View file

@ -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

15
toolkits/zendesk/arcade_zendesk/__init__.py Normal file
View file

@ -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",
]

27
toolkits/zendesk/arcade_zendesk/enums.py Normal file
View file

@ -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"

15
toolkits/zendesk/arcade_zendesk/tools/__init__.py Normal file
View file

@ -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",
]

227
toolkits/zendesk/arcade_zendesk/tools/search_articles.py Normal file
View file

@ -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

414
toolkits/zendesk/arcade_zendesk/tools/tickets.py Normal file
View file

@ -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

216
toolkits/zendesk/arcade_zendesk/utils.py Normal file
View file

@ -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

360
toolkits/zendesk/evals/eval_articles.py Normal file
View file

@ -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

631
toolkits/zendesk/evals/eval_tickets.py Normal file
View file

@ -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

59
toolkits/zendesk/pyproject.toml Normal file
View file

@ -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",]

0
toolkits/zendesk/tests/__init__.py Normal file
View file

84
toolkits/zendesk/tests/conftest.py Normal file
View file

@ -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": "<p>To reset your password, follow these steps:</p>"
"<ol><li>Click forgot password</li><li>Enter your email</li></ol>",
"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

423
toolkits/zendesk/tests/test_search_articles.py Normal file
View file

@ -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": "<h1>Header</h1><p>Paragraph with <strong>bold</strong> and "
"<em>italic</em>.</p><br/><div>Div content</div>",
"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

526
toolkits/zendesk/tests/test_tickets.py Normal file
View file

@ -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]

291
toolkits/zendesk/tests/test_utils.py Normal file
View file

@ -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 = "<p>Hello <strong>World</strong></p>"
assert clean_html_text(html) == "Hello World"
def test_clean_complex_html(self):
"""Test cleaning complex HTML with multiple tags."""
html = """
<h1>Title</h1>
<p>Paragraph with <em>emphasis</em> and <strong>bold</strong>.</p>
<ul>
<li>Item 1</li>
<li>Item 2</li>
</ul>
<div class="footer">Footer content</div>
"""
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 "<h1>" not in cleaned
assert "<li>" not in cleaned
def test_clean_html_with_special_chars(self):
"""Test cleaning HTML with special characters."""
html = "<p>Price: &pound;100 &amp; &euro;120</p>"
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, ""),
("", ""),
(" ", ""),
("<p></p>", ""),
("<p> </p>", ""),
],
)
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 = "<p>Line 1</p><p>Line 2</p>"
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 = "<h1>Article Title</h1><p>Article content with <strong>formatting</strong>.</p>"
result = process_article_body(body)
assert "Article Title" in result
assert "Article content with formatting" in result
assert "<h1>" not in result
assert "<strong>" not in result
def test_process_body_with_truncation(self):
"""Test processing body with max length."""
body = "<p>" + "Long content " * 50 + "</p>"
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),
("<p>Short</p>", 100, "Short"),
(
"<p></p>",
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": "<p>Content 1</p>",
"url": "https://example.com/1",
},
{
"id": 2,
"title": "Article 2",
"body": "<p>Content 2</p>",
"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": "<p>Content 1</p>",
"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": "<p>" + "Long content " * 100 + "</p>",
}
]
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": "<p>Content</p>",
"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