Add Confluence toolkit (#406)

12 tools for the new Confluence toolkit.

| Name | Description |

|----------------------------------|------------------------------------------------------------------------------------|
| Confluence.CreatePage | Create a new page at the root of the given
space. |
| Confluence.UpdatePageContent | Update a page's content. |
| Confluence.RenamePage | Rename a page by changing its title. |
| Confluence.GetPage | Retrieve a SINGLE page's content by its ID or
title. |
| Confluence.GetPagesById | Get the content of MULTIPLE pages by their
ID in a single efficient request. |
| Confluence.ListPages | Get the content of multiple pages by their ID.
|
| Confluence.ListAttachments | List attachments in a workspace. |
| Confluence.GetAttachmentsForPage | Get attachments for a page by its
ID or title. |
| Confluence.SearchContent | Search for content in Confluence. |
| Confluence.GetSpace | Get the details of a space by its ID or key. |
| Confluence.ListSpaces | List all spaces sorted by name in ascending
order. |
| Confluence.GetSpaceHierarchy | Retrieve the full hierarchical
structure of a Confluence space as a tree structure.|

### Confluence Clients
Confluence has deprecated most of their V1 endpoints, so most of the
tools use V2. However, we still need a V1 client to support search. The
V1 search API has not been deprecated yet, because there is no V2
equivalent. But we need to be aware in the future for when this
deprecation happens.


### Future work
* Content of pages are returned in the Confluence `storage` format. This
is the format that Confluence uses to store pages internally. We should
understand the storage format more deeply, and write utility function to
transform this format into plain text.
* Better protections against extremely large pages. I've tested up to
6,000 word pages.
* Tools for blog posts
* Tool for getting the children of a page. (`get_space_hierarchy` will
suffice for now)
* Allow for numerical titles
This commit is contained in:
Eric Gustin 2025-05-22 12:18:19 -08:00 committed by GitHub
parent 24758ad35a
commit 6bc6f0c574
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 2406 additions and 0 deletions

View file

@ -1,5 +1,6 @@
arcade-asana
arcade-code-sandbox
arcade-confluence
arcade-dropbox
arcade-github
arcade-google

View file

@ -0,0 +1,18 @@
files: ^./
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

View file

@ -0,0 +1,44 @@
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",
]
[lint.per-file-ignores]
"**/tests/*" = ["S101"]
[format]
preview = true
skip-magic-trailing-comma = false

View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025, Arcade
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

View file

@ -0,0 +1,58 @@
.PHONY: help
help:
@echo "🛠️ confluence 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 poetry environment and install the pre-commit hooks
@echo "📦 Checking if Poetry is installed"
@if ! command -v poetry >/dev/null 2>&1; then \
echo "📦 Poetry not found. Checking if pip is available"; \
if ! command -v pip >/dev/null 2>&1; then \
echo "❌ pip is not installed. Please install pip first."; \
exit 1; \
fi; \
echo "📦 Installing Poetry with pip"; \
pip install poetry==1.8.5; \
else \
echo "📦 Poetry is already installed"; \
fi
@echo "🚀 Installing package in development mode with all extras"
poetry install --all-extras
.PHONY: build
build: clean-build ## Build wheel file using poetry
@echo "🚀 Creating wheel file"
poetry 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"
@poetry run pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
.PHONY: coverage
coverage: ## Generate coverage report
@echo "coverage report"
coverage report
@echo "Generating coverage report"
coverage html
.PHONY: bump-version
bump-version: ## Bump the version in the pyproject.toml file
@echo "🚀 Bumping version in pyproject.toml"
poetry version patch
.PHONY: check
check: ## Run code quality tools.
@echo "🚀 Checking Poetry lock file consistency with 'pyproject.toml': Running poetry check"
@poetry check
@echo "🚀 Linting code: Running pre-commit"
@poetry run pre-commit run -a
@echo "🚀 Static type checking: Running mypy"
@poetry run mypy --config-file=pyproject.toml

View file

@ -0,0 +1,614 @@
from enum import Enum
from typing import Any
from urllib.parse import parse_qs, urlparse
import httpx
from arcade.sdk.errors import ToolExecutionError
from arcade_confluence.enums import BodyFormat, PageUpdateMode
from arcade_confluence.utils import (
build_child_url,
build_hierarchy,
remove_none_values,
)
class ConfluenceAPIVersion(str, Enum):
V1 = "wiki/rest/api"
V2 = "wiki/api/v2"
class ConfluenceClient:
ACCESSIBLE_RESOURCES_URL = "https://api.atlassian.com/oauth/token/accessible-resources"
BASE_URL = "https://api.atlassian.com/ex/confluence"
def __init__(self, token: str, api_version: ConfluenceAPIVersion):
self.token = token
self.cloud_id = self._get_cloud_id()
self.api_version = api_version.value
def _get_cloud_id(self) -> str:
"""
Fetch the cloudId for <workspace_name>.atlassian.net
using the OAuth2 3LO accessible-resources endpoint.
For details on why this is necessary, see: https://developer.atlassian.com/cloud/oauth/getting-started/making-calls-to-api
"""
headers = {"Authorization": f"Bearer {self.token}"}
resp = httpx.get(self.ACCESSIBLE_RESOURCES_URL, headers=headers)
resp.raise_for_status()
resp_json = resp.json()
if len(resp_json) == 0:
raise ToolExecutionError(message="No workspaces found for the authenticated user.")
return resp_json[0].get("id") # type: ignore[no-any-return]
async def request(self, method: str, path: str, **kwargs: Any) -> Any:
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {self.token}",
}
async with httpx.AsyncClient() as client:
response = await client.request(
method,
f"{self.BASE_URL}/{self.cloud_id}/{self.api_version}/{path.lstrip('/')}",
headers=headers,
**kwargs,
)
response.raise_for_status()
return response.json()
async def get(self, path: str, **kwargs: Any) -> Any:
return await self.request("GET", path, **kwargs)
async def post(self, path: str, **kwargs: Any) -> Any:
return await self.request("POST", path, **kwargs)
async def put(self, path: str, **kwargs: Any) -> Any:
return await self.request("PUT", path, **kwargs)
class ConfluenceClientV1(ConfluenceClient):
def __init__(self, token: str):
super().__init__(token, api_version=ConfluenceAPIVersion.V1)
def _build_query_cql(self, query: str, enable_fuzzy: bool) -> str:
"""Build CQL for a single query (term or phrase).
Args:
query: The search query (single word term or multi-word phrase)
enable_fuzzy: Whether to enable fuzzy matching for single terms
Returns:
CQL string for the query
"""
query = query.strip()
if not query:
return ""
# For phrases (multiple words), don't use fuzzy matching
if " " in query:
return f'(text ~ "{query}" OR title ~ "{query}" OR space.title ~ "{query}")'
else:
# For single terms, optionally use fuzzy matching
term_suffix = "~" if enable_fuzzy else ""
return f'(text ~ "{query}{term_suffix}" OR title ~ "{query}{term_suffix}" OR space.title ~ "{query}{term_suffix}")' # noqa: E501
def _build_and_cql(self, queries: list[str], enable_fuzzy: bool) -> str:
"""Build CQL for queries that must ALL be present (AND logic).
Args:
queries: List of queries that must all be present
enable_fuzzy: Whether to enable fuzzy matching for single terms
Returns:
CQL string with AND logic
"""
and_parts = []
for query in queries:
query_cql = self._build_query_cql(query, enable_fuzzy)
if query_cql:
and_parts.append(query_cql)
if not and_parts:
return ""
return f"({' AND '.join(and_parts)})"
def _build_or_cql(self, queries: list[str], enable_fuzzy: bool) -> str:
"""Build CQL for queries where ANY can be present (OR logic).
Args:
queries: List of queries where any can be present
enable_fuzzy: Whether to enable fuzzy matching for single terms
Returns:
CQL string with OR logic
"""
or_parts = []
for query in queries:
query_cql = self._build_query_cql(query, enable_fuzzy)
if query_cql:
or_parts.append(query_cql)
if not or_parts:
return ""
return f"({' OR '.join(or_parts)})"
def construct_cql(
self,
must_contain_all: list[str] | None,
can_contain_any: list[str] | None,
enable_fuzzy: bool = False,
) -> str:
"""Construct CQL query with AND/OR logic.
Learn about advanced searching using CQL here: https://developer.atlassian.com/cloud/confluence/advanced-searching-using-cql/
Args:
must_contain_all: Queries that must ALL be present (AND logic)
can_contain_any: Queries where ANY can be present (OR logic)
enable_fuzzy: Whether to enable fuzzy matching for single terms
Returns:
CQL query string
Raises:
ToolExecutionError: If no search parameters are provided
"""
cql_parts = []
# Handle must_contain_all (AND logic)
if must_contain_all:
and_cql = self._build_and_cql(must_contain_all, enable_fuzzy)
if and_cql:
cql_parts.append(and_cql)
# Handle can_contain_any (OR logic)
if can_contain_any:
or_cql = self._build_or_cql(can_contain_any, enable_fuzzy)
if or_cql:
cql_parts.append(or_cql)
# If there's only one part, return it
if len(cql_parts) == 1:
return cql_parts[0]
# AND the must_contain_all with the can_contain_any
if len(cql_parts) > 1:
return f"({' AND '.join(cql_parts)})"
raise ToolExecutionError(message="At least one search parameter must be provided")
def transform_search_content_response(
self, response: dict[str, Any]
) -> dict[str, list[dict[str, Any]]]:
"""
Transform the response from the GET /search endpoint by converting relative webui paths
to absolute URLs using the base URL from the response.
"""
base_url = response.get("_links", {}).get("base", "")
transformed_results = []
for result in response.get("results", []):
content = result.get("content", {})
transformed_result = {
"id": content.get("id"),
"title": content.get("title"),
"type": content.get("type"),
"status": content.get("status"),
"excerpt": result.get("excerpt"),
"url": f"{base_url}{result.get('url')}",
}
transformed_results.append(transformed_result)
return {"results": transformed_results}
class ConfluenceClientV2(ConfluenceClient):
def __init__(self, token: str):
super().__init__(token, api_version=ConfluenceAPIVersion.V2)
def _transform_links(
self, response: dict[str, Any], base_url: str | None = None
) -> dict[str, Any]:
"""
Transform the links in a page response by converting relative URLs to absolute URLs.
Args:
response: A page object from the API
base_url: The base URL to use for the transformation
Returns:
The transformed response
"""
result = response.copy()
if "_links" in result:
base_url = base_url or result["_links"].get("base", "")
webui_path = result["_links"].get("webui", "")
result["url"] = f"{base_url}{webui_path}"
del result["_links"]
return result
def transform_get_spaces_response(
self, response: dict[str, Any]
) -> dict[str, list[dict[str, Any]]]:
"""
Transform the response from the GET /spaces endpoint by converting relative webui paths
to absolute URLs using the base URL from the response.
"""
pagination_token = parse_qs(urlparse(response.get("_links", {}).get("next", "")).query).get(
"cursor", [None]
)[0]
base_url = response.get("_links", {}).get("base", "")
results = response.get("results", [])
transformed_results = []
for space in results:
space_copy = space.copy()
if "_links" in space_copy and "webui" in space_copy["_links"]:
webui_path = space_copy["_links"]["webui"]
space_copy["url"] = base_url + webui_path
del space_copy["_links"]
transformed_results.append(space_copy)
results = {"spaces": transformed_results, "pagination_token": pagination_token}
return remove_none_values(results)
def transform_list_pages_response(self, response: dict[str, Any]) -> dict[str, Any]:
"""Transform the response from the GET /pages endpoint."""
pagination_token = parse_qs(urlparse(response.get("_links", {}).get("next", "")).query).get(
"cursor", [None]
)[0]
base_url = response.get("_links", {}).get("base", "")
pages = [self._transform_links(page, base_url) for page in response["results"]]
results = {"pages": pages, "pagination_token": pagination_token}
return remove_none_values(results)
def transform_get_multiple_pages_response(
self, response: dict[str, Any]
) -> dict[str, list[dict[str, Any]]]:
"""Transform the response from the GET /pages endpoint."""
base_url = response.get("_links", {}).get("base", "")
pages = [self._transform_links(page, base_url) for page in response["results"]]
return {"pages": pages}
def transform_space_response(
self, response: dict[str, Any], base_url: str | None = None
) -> dict[str, dict[str, Any]]:
"""Transform API responses that return a space object."""
return {"space": self._transform_links(response, base_url)}
def transform_page_response(self, response: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""Transform API responses that return a page object."""
return {"page": self._transform_links(response)}
def transform_get_attachments_response(self, response: dict[str, Any]) -> dict[str, Any]:
"""Transform the response from the GET /pages/{id}/attachments endpoint."""
pagination_token = parse_qs(urlparse(response.get("_links", {}).get("next", "")).query).get(
"cursor", [None]
)[0]
base_url = response.get("_links", {}).get("base", "")
attachments = []
for attachment in response["results"]:
result = attachment.copy()
if "_links" in result:
webui_path = result["_links"].get("webui", "")
download_path = result["_links"].get("download", "")
result["url"] = f"{base_url}{webui_path}"
result["download_link"] = f"{base_url}{download_path}"
del result["_links"]
del result["webuiLink"]
del result["downloadLink"]
del result["version"]
attachments.append(result)
return {"attachments": attachments, "pagination_token": pagination_token}
def prepare_update_page_payload(
self,
page_id: str,
status: str,
title: str,
body_representation: str,
body_value: str,
version_number: int,
version_message: str,
) -> dict[str, Any]:
"""Prepare a payload for the PUT /pages/{id} endpoint."""
return {
"id": page_id,
"status": status,
"title": title,
"body": {
"representation": body_representation,
"value": body_value,
},
"version": {
"number": version_number,
"message": version_message,
},
}
def prepare_update_page_content_payload(
self,
content: str,
update_mode: PageUpdateMode,
old_content: str,
page_id: str,
status: str,
title: str,
body_representation: BodyFormat,
old_version_number: int,
) -> dict[str, Any]:
"""Prepare a payload for when updating the content of a page
Args:
content: The content to update the page with
update_mode: The mode of update to use
old_content: The content of the page before the update
page_id: The ID of the page to update
status: The status of the page
title: The title of the page
body_representation: The format that the body (content) is in
old_version_number: The version number of the page before the update
Returns:
A payload for the PUT /pages/{id} endpoint's json body
"""
updated_content = ""
updated_message = ""
if update_mode == PageUpdateMode.APPEND:
updated_content = f"{old_content}<br/>{content}"
updated_message = "Append content to the page"
elif update_mode == PageUpdateMode.PREPEND:
updated_content = f"{content}<br/>{old_content}"
updated_message = "Prepend content to the page"
elif update_mode == PageUpdateMode.REPLACE:
updated_content = content
updated_message = "Replace the page content"
payload = self.prepare_update_page_payload(
page_id=page_id,
status=status,
title=title,
body_representation=body_representation.to_api_value(),
body_value=updated_content,
version_number=old_version_number + 1,
version_message=updated_message,
)
return payload
async def get_root_pages_in_space(self, space_id: str) -> dict[str, Any]:
"""
Get the root pages in a space.
Requires Confluence scope 'read:page:confluence'
"""
params = {
"depth": "root",
"limit": 250,
}
pages = await self.get(f"spaces/{space_id}/pages", params=params)
base_url = pages.get("_links", {}).get("base", "")
return {"pages": [self._transform_links(page, base_url) for page in pages["results"]]}
async def get_space_homepage(self, space_id: str) -> dict[str, Any]:
"""
Get the homepage of a space.
Requires Confluence scope 'read:page:confluence'
"""
root_pages = await self.get_root_pages_in_space(space_id)
for page in root_pages["pages"]:
if page.get("url", "").endswith("overview"):
return self._transform_links(page)
raise ToolExecutionError(message="No homepage found for space.")
async def get_page_by_id(
self, page_id: str, content_format: BodyFormat = BodyFormat.STORAGE
) -> dict[str, Any]:
"""Get a page by its ID.
Requires Confluence scope 'read:page:confluence'
Args:
page_id: The ID of the page to get
content_format: The format of the page content
Returns:
The page object
"""
params = remove_none_values({
"body-format": content_format.to_api_value(),
})
try:
page = await self.get(f"pages/{page_id}", params=params)
except httpx.HTTPStatusError as e:
# If the page is not found, return an empty page object
if e.response.status_code in [400, 404]:
return self.transform_page_response({})
raise
return self.transform_page_response(page)
async def get_page_by_title(
self, page_title: str, content_format: BodyFormat = BodyFormat.STORAGE
) -> dict[str, Any]:
"""Get a page by its title.
Requires Confluence scope 'read:page:confluence'
Args:
page_title: The title of the page to get
content_format: The format of the page content
Returns:
The page object
"""
params = {
"title": page_title,
"body-format": content_format.to_api_value(),
}
response = await self.get("pages", params=params)
pages = response.get("results", [])
if not pages:
# If the page is not found, return an empty page object
return self.transform_page_response({})
return self.transform_page_response(pages[0])
async def get_space_by_id(self, space_id: str) -> dict[str, Any]:
"""Get a space by its ID.
Requires Confluence scope 'read:space:confluence'
Args:
space_id: The ID of the space to get
Returns:
The space object
"""
space = await self.get(f"spaces/{space_id}")
return self.transform_space_response(space)
async def get_space_by_key(self, space_key: str) -> dict[str, Any]:
"""Get a space by its key.
Requires Confluence scope 'read:space:confluence'
Args:
space_key: The key of the space to get
Returns:
The space object
"""
response = await self.get("spaces", params={"keys": [space_key]})
base_url = response.get("_links", {}).get("base", "")
spaces = response.get("results", [])
if not spaces:
raise ToolExecutionError(message=f"No space found with key: '{space_key}'")
return self.transform_space_response(spaces[0], base_url=base_url)
async def get_space(self, space_identifier: str) -> dict[str, Any]:
"""Get a space from its identifier.
Requires Confluence scope 'read:space:confluence'
Args:
space_identifier: The identifier of the space to get. Can be a space's ID or key.
"""
return (
await self.get_space_by_id(space_identifier)
if space_identifier.isdigit()
else await self.get_space_by_key(space_identifier)
)
async def get_page_id(self, page_identifier: str) -> str:
"""Get the ID of a page from its identifier.
Args:
page_identifier: The identifier of the page to get. Can be a page's ID or title.
Returns:
The ID of the page
"""
if page_identifier.isdigit():
page_id = page_identifier
else:
page = await self.get_page_by_title(page_identifier)
page_id = page.get("page", {}).get("id")
if not page_id:
raise ToolExecutionError(message=f"No page found with identifier: '{page_identifier}'")
return page_id
async def get_space_id(self, space_identifier: str) -> str:
"""Get the ID of a space from its identifier.
Args:
space_identifier: The identifier of the space to get. Can be a space's ID or title.
"""
if space_identifier.isdigit():
space_id = space_identifier
else:
space = await self.get_space_by_key(space_identifier)
space_id = space.get("space", {}).get("id")
return space_id
def create_space_tree(self, space: dict) -> dict:
"""Create the initial tree structure for a space.
Args:
space: The transformed space object
Returns:
A dictionary representing the root of the space hierarchy tree without any children
"""
space_internal = space.get("space", {})
return {
"key": space_internal.get("key"),
"id": space_internal.get("id"),
"type": "space",
"url": space_internal.get("url"),
"children": [],
}
def convert_root_pages_to_tree_nodes(self, pages: list) -> list:
"""Convert root pages of a space to tree nodes.
Args:
pages: List of page objects from the API
Returns:
A list of tree nodes representing the root pages
"""
return [
{
"title": page.get("title"),
"id": page.get("id"),
"type": "page",
"url": page.get("url"),
"children": [],
}
for page in pages
]
async def process_page_descendants(self, root_children: list, base_url: str) -> None:
"""Process descendants for each page and build the hierarchy.
Args:
root_children: The root children of the space
base_url: The base URL for the Confluence space
Returns:
None (modifies root_children in place)
"""
descendent_params = {"limit": 250, "depth": 5}
for i, child in enumerate(root_children):
page_id = child["id"]
descendants = await self.get(f"pages/{page_id}/descendants", params=descendent_params)
# Transform descendants into our desired format
transformed_children = []
for descendant in descendants.get("results", []):
transformed_child = {
"title": descendant.get("title"),
"id": descendant.get("id"),
"type": descendant.get("type"),
"parent_id": page_id
if descendant.get("parentId") is None
else descendant.get("parentId"),
"parent_type": descendant.get("parentType", "TODO"),
"url": build_child_url(base_url, descendant),
"children": [],
"status": descendant.get("status"),
}
transformed_children.append(transformed_child)
# Build the hierarchy for the current root page
build_hierarchy(transformed_children, page_id, root_children[i])

View file

@ -0,0 +1,63 @@
from enum import Enum
class BodyFormat(str, Enum):
STORAGE = "storage" # Storage representation for editing, with relative urls in the markup
def to_api_value(self) -> str:
mapping = {
BodyFormat.STORAGE: "storage",
}
return mapping.get(self, BodyFormat.STORAGE.value)
class PageUpdateMode(str, Enum):
"""The mode of update for a page"""
PREPEND = "prepend" # Add content to the beginning of the page.
APPEND = "append" # Add content to the end of the page.
REPLACE = "replace" # Replace the entire page with the new content.
class PageSortOrder(str, Enum):
"""The order of the pages to sort by"""
ID_ASCENDING = "id-ascending"
ID_DESCENDING = "id-descending"
TITLE_ASCENDING = "title-ascending"
TITLE_DESCENDING = "title-descending"
CREATED_DATE_ASCENDING = "created-date-oldest-to-newest"
CREATED_DATE_DESCENDING = "created-date-newest-to-oldest"
MODIFIED_DATE_ASCENDING = "modified-date-oldest-to-newest"
MODIFIED_DATE_DESCENDING = "modified-date-newest-to-oldest"
def to_api_value(self) -> str:
mapping = {
PageSortOrder.ID_ASCENDING: "id",
PageSortOrder.ID_DESCENDING: "-id",
PageSortOrder.TITLE_ASCENDING: "title",
PageSortOrder.TITLE_DESCENDING: "-title",
PageSortOrder.CREATED_DATE_ASCENDING: "created-date",
PageSortOrder.CREATED_DATE_DESCENDING: "-created-date",
PageSortOrder.MODIFIED_DATE_ASCENDING: "modified-date",
PageSortOrder.MODIFIED_DATE_DESCENDING: "-modified-date",
}
return mapping.get(self, PageSortOrder.CREATED_DATE_DESCENDING.value)
class AttachmentSortOrder(str, Enum):
"""The order of the attachments to sort by"""
CREATED_DATE_ASCENDING = "created-date-oldest-to-newest"
CREATED_DATE_DESCENDING = "created-date-newest-to-oldest"
MODIFIED_DATE_ASCENDING = "modified-date-oldest-to-newest"
MODIFIED_DATE_DESCENDING = "modified-date-newest-to-oldest"
def to_api_value(self) -> str:
mapping = {
AttachmentSortOrder.CREATED_DATE_ASCENDING: "created-date",
AttachmentSortOrder.CREATED_DATE_DESCENDING: "-created-date",
AttachmentSortOrder.MODIFIED_DATE_ASCENDING: "modified-date",
AttachmentSortOrder.MODIFIED_DATE_DESCENDING: "-modified-date",
}
return mapping.get(self, AttachmentSortOrder.CREATED_DATE_DESCENDING.value)

View file

@ -0,0 +1,30 @@
from arcade_confluence.tools.attachment import get_attachments_for_page, list_attachments
from arcade_confluence.tools.page import (
create_page,
get_page,
get_pages_by_id,
list_pages,
rename_page,
update_page_content,
)
from arcade_confluence.tools.search import search_content
from arcade_confluence.tools.space import get_space, get_space_hierarchy, list_spaces
__all__ = [
# Attachment
"get_attachments_for_page",
"list_attachments",
# Page
"create_page",
"get_pages_by_id",
"get_page",
"list_pages",
"rename_page",
"update_page_content",
# Search
"search_content",
# Space
"get_space",
"get_space_hierarchy",
"list_spaces",
]

View file

@ -0,0 +1,69 @@
from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Atlassian
from arcade_confluence.client import ConfluenceClientV2
from arcade_confluence.enums import AttachmentSortOrder
from arcade_confluence.utils import remove_none_values
@tool(
requires_auth=Atlassian(
scopes=["read:attachment:confluence"],
)
)
async def list_attachments(
context: ToolContext,
sort_order: Annotated[
AttachmentSortOrder,
"The order of the attachments to sort by. Defaults to created-date-newest-to-oldest",
] = AttachmentSortOrder.CREATED_DATE_DESCENDING,
limit: Annotated[
int, "The maximum number of attachments to return. Defaults to 25. Max is 250"
] = 25,
pagination_token: Annotated[
str | None,
"The pagination token to use for the next page of results",
] = None,
) -> Annotated[dict, "The attachments"]:
"""List attachments in a workspace"""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
params = remove_none_values({
"sort": sort_order.to_api_value(),
"limit": max(1, min(limit, 250)),
"cursor": pagination_token,
})
attachments = await client.get("attachments", params=params)
return client.transform_get_attachments_response(attachments)
@tool(
requires_auth=Atlassian(
scopes=["read:attachment:confluence"],
)
)
async def get_attachments_for_page(
context: ToolContext,
page_identifier: Annotated[str, "The ID or title of the page to get attachments for"],
limit: Annotated[
int, "The maximum number of attachments to return. Defaults to 25. Max is 250"
] = 25,
pagination_token: Annotated[
str | None,
"The pagination token to use for the next page of results",
] = None,
) -> Annotated[dict, "The attachments"]:
"""Get attachments for a page by its ID or title.
If a page title is provided, then the first page with an exact matching title will be returned.
"""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
page_id = await client.get_page_id(page_identifier)
params = remove_none_values({
"limit": max(1, min(limit, 250)),
"cursor": pagination_token,
})
attachments = await client.get(f"pages/{page_id}/attachments", params=params)
return client.transform_get_attachments_response(attachments)

View file

@ -0,0 +1,236 @@
from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Atlassian
from arcade.sdk.errors import ToolExecutionError
from arcade_confluence.client import ConfluenceClientV2
from arcade_confluence.enums import BodyFormat, PageSortOrder, PageUpdateMode
from arcade_confluence.utils import remove_none_values, validate_ids
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence", "write:page:confluence"],
)
)
async def create_page(
context: ToolContext,
space_identifier: Annotated[str, "The ID or title of the space to create the page in"],
title: Annotated[str, "The title of the page"],
content: Annotated[str, "The content of the page. Only plain text is supported"],
parent_id: Annotated[
str | None,
"The ID of the parent. If not provided, the page will be created at the root of the space.",
] = None,
is_private: Annotated[
bool,
"If true, then only the user who creates this page will be able to see it. "
"Defaults to False",
] = False,
is_draft: Annotated[
bool,
"If true, then the page will be created as a draft. Defaults to False",
] = False,
) -> Annotated[dict, "The page"]:
"""Create a new page at the root of the given space."""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
space_id = await client.get_space_id(space_identifier)
parent_id = parent_id or (await client.get_space_homepage(space_id)).get("id")
params = remove_none_values({
"root-level": False,
"private": is_private,
})
body = remove_none_values({
"spaceId": space_id,
"status": "draft" if is_draft else None,
"parentId": parent_id,
"title": title,
"body": {
"storage": {
"value": content,
"representation": BodyFormat.STORAGE.to_api_value(),
}
},
})
page = await client.post("pages", params=params, json=body)
return client.transform_page_response(page)
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence", "write:page:confluence"],
)
)
async def update_page_content(
context: ToolContext,
page_identifier: Annotated[
str, "The ID or title of the page to update. Numerical titles are NOT supported."
],
content: Annotated[str, "The content of the page. Only plain text is supported"],
update_mode: Annotated[
PageUpdateMode,
"The mode of update. Defaults to 'append'.",
] = PageUpdateMode.APPEND,
) -> Annotated[dict, "The page"]:
"""Update a page's content."""
# Get the page to update
client = ConfluenceClientV2(context.get_auth_token_or_empty())
page_id = await client.get_page_id(page_identifier)
page = await get_page(context, page_id)
if not page.get("page"):
raise ToolExecutionError(message=f"No page found with identifier: '{page_identifier}'")
status = page.get("page", {}).get("status", "current")
title = page.get("page", {}).get("title", "Untitled page")
body = page.get("page", {}).get("body", {})
old_content = body.get(BodyFormat.STORAGE, {}).get("value", "")
old_version_number = page.get("page", {}).get("version", {}).get("number", 0)
# Update the page content
payload = client.prepare_update_page_content_payload(
content=content,
update_mode=update_mode,
old_content=old_content,
page_id=page_id,
status=status,
title=title,
body_representation=BodyFormat.STORAGE,
old_version_number=old_version_number,
)
updated_page = await client.put(f"pages/{page_id}", json=payload)
return client.transform_page_response(updated_page)
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence", "write:page:confluence"],
)
)
async def rename_page(
context: ToolContext,
page_identifier: Annotated[
str, "The ID or title of the page to rename. Numerical titles are NOT supported."
],
title: Annotated[str, "The title of the page"],
) -> Annotated[dict, "The page"]:
"""Rename a page by changing its title."""
# Get the page to rename
client = ConfluenceClientV2(context.get_auth_token_or_empty())
page_id = await client.get_page_id(page_identifier)
page = await get_page(context, page_id)
if not page.get("page"):
raise ToolExecutionError(message=f"No page found with identifier: '{page_identifier}'")
status = page.get("page", {}).get("status", "current")
content = page.get("page", {}).get("body", {}).get(BodyFormat.STORAGE, {}).get("value", "")
old_version_number = page.get("page", {}).get("version", {}).get("number", 0)
# Rename the page
payload = client.prepare_update_page_payload(
page_id=page_id,
status=status,
title=title,
body_representation=BodyFormat.STORAGE,
body_value=content,
version_number=old_version_number + 1,
version_message="Rename the page",
)
updated_page = await client.put(f"pages/{page_id}", json=payload)
return client.transform_page_response(updated_page)
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence"],
)
)
async def get_page(
context: ToolContext,
page_identifier: Annotated[
str, "Can be a page's ID or title. Numerical titles are NOT supported."
],
) -> Annotated[dict, "The page"]:
"""Retrieve a SINGLE page's content by its ID or title.
If a title is provided, then the first page with an exact matching title will be returned.
IMPORTANT: For retrieving MULTIPLE pages, use `get_pages_by_id` instead
for a massive performance and efficiency boost. If you call this function multiple times
instead of using `get_pages_by_id`, then the universe will explode.
"""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
if page_identifier.isdigit():
return await client.get_page_by_id(page_identifier, BodyFormat.STORAGE)
else:
return await client.get_page_by_title(page_identifier, BodyFormat.STORAGE)
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence"],
)
)
async def get_pages_by_id(
context: ToolContext,
page_ids: Annotated[
list[str],
"The IDs of the pages to get. IDs are numeric. Titles of pages are NOT supported. "
"Maximum of 250 page ids supported.",
],
) -> Annotated[dict, "The pages"]:
"""Get the content of MULTIPLE pages by their ID in a single efficient request.
IMPORTANT: Always use this function when you need to retrieve content from more than one page,
rather than making multiple separate calls to get_page, because this function is significantly
more efficient than calling get_page multiple times.
"""
validate_ids(page_ids, max_length=250)
client = ConfluenceClientV2(context.get_auth_token_or_empty())
pages = await client.get(
"pages", params={"id": page_ids, "body-format": BodyFormat.STORAGE.to_api_value()}
)
return client.transform_get_multiple_pages_response(pages)
@tool(
requires_auth=Atlassian(
scopes=["read:page:confluence"],
)
)
async def list_pages(
context: ToolContext,
space_ids: Annotated[
list[str] | None,
"Restrict the response to only include pages in these spaces. "
"Only space IDs are supported. Titles of spaces are NOT supported. "
"If not provided, then no restriction is applied. "
"Maximum of 100 space ids supported.",
] = None,
sort_by: Annotated[
PageSortOrder,
"The order of the pages to sort by. Defaults to created-date-newest-to-oldest",
] = PageSortOrder.CREATED_DATE_DESCENDING,
limit: Annotated[int, "The maximum number of pages to return. Defaults to 25. Max is 250"] = 25,
pagination_token: Annotated[
str | None,
"The pagination token to use for the next page of results",
] = None,
) -> Annotated[dict, "The pages"]:
"""Get the content of multiple pages by their ID"""
validate_ids(space_ids, max_length=100)
limit = max(1, min(limit, 250))
client = ConfluenceClientV2(context.get_auth_token_or_empty())
params = remove_none_values({
"space-id": space_ids,
"sort": sort_by.to_api_value(),
"body-format": BodyFormat.STORAGE.to_api_value(),
"limit": limit,
"cursor": pagination_token,
})
pages = await client.get("pages", params=params)
return client.transform_list_pages_response(pages)

View file

@ -0,0 +1,53 @@
from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Atlassian
from arcade_confluence.client import ConfluenceClientV1
@tool(
requires_auth=Atlassian(
scopes=["search:confluence"],
)
)
async def search_content(
context: ToolContext,
must_contain_all: Annotated[
list[str] | None,
"Words/phrases that content MUST contain (AND logic). Each item can be:\n"
"- Single word: 'banana' - content must contain this word\n"
"- Multi-word phrase: 'How to' - content must contain all these words (in any order)\n"
"- All items in this list must be present for content to match\n"
"- Example: ['banana', 'apple'] finds content containing BOTH 'banana' AND 'apple'",
] = None,
can_contain_any: Annotated[
list[str] | None,
"Words/phrases where content can contain ANY of these (OR logic). Each item can be:\n"
"- Single word: 'project' - content containing this word will match\n"
"- Multi-word phrase: 'pen & paper' - content containing all these words will match\n"
"- Content matching ANY item in this list will be included\n"
"- Example: ['project', 'documentation'] finds content with 'project' OR 'documentation'",
] = None,
enable_fuzzy: Annotated[
bool,
"Enable fuzzy matching to find similar terms (e.g. 'roam' will find 'foam'). "
"Defaults to True",
] = True,
limit: Annotated[int, "Maximum number of results to return (1-100). Defaults to 25"] = 25,
) -> Annotated[dict, "Search results containing content items matching the criteria"]:
"""Search for content in Confluence.
The search is performed across all content in the authenticated user's Confluence workspace.
All search terms in Confluence are case insensitive.
You can use the parameters in different ways:
- must_contain_all: For AND logic - content must contain ALL of these
- can_contain_any: For OR logic - content can contain ANY of these
- Combine them: must_contain_all=['banana'] AND can_contain_any=['database', 'guide']
"""
client = ConfluenceClientV1(context.get_auth_token_or_empty())
cql = client.construct_cql(must_contain_all, can_contain_any, enable_fuzzy)
response = await client.get("search", params={"cql": cql, "limit": max(1, min(limit, 100))})
return client.transform_search_content_response(response)

View file

@ -0,0 +1,95 @@
import re
from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Atlassian
from arcade_confluence.client import ConfluenceClientV2
from arcade_confluence.utils import remove_none_values
@tool(
requires_auth=Atlassian(
scopes=["read:space:confluence"],
)
)
async def get_space(
context: ToolContext,
space_identifier: Annotated[
str, "Can be a space's ID or key. Numerical keys are NOT supported"
],
) -> Annotated[dict, "The space"]:
"""Get the details of a space by its ID or key."""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
if space_identifier.isdigit():
return await client.get_space_by_id(space_identifier)
else:
return await client.get_space_by_key(space_identifier)
@tool(
requires_auth=Atlassian(
scopes=["read:space:confluence"],
)
)
async def list_spaces(
context: ToolContext,
limit: Annotated[
int, "The maximum number of spaces to return. Defaults to 25. Max is 250"
] = 25,
pagination_token: Annotated[
str | None, "The pagination token to use for the next page of results"
] = None,
) -> Annotated[dict, "The spaces"]:
"""List all spaces sorted by name in ascending order."""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
params = {"limit": max(1, min(limit, 250)), "sort": "name", "cursor": pagination_token}
params = remove_none_values(params)
spaces = await client.get("spaces", params=params)
return client.transform_get_spaces_response(spaces)
@tool(
requires_auth=Atlassian(
scopes=[
"read:page:confluence", # needed for getting the space's root pages
"read:space:confluence", # needed for when a space key is provided
"read:hierarchical-content:confluence", # needed for getting the descendents of a page
],
)
)
async def get_space_hierarchy(
context: ToolContext,
space_identifier: Annotated[
str, "Can be a space's ID or key. Numerical keys are NOT supported"
],
) -> Annotated[dict, "The space hierarchy"]:
"""Retrieve the full hierarchical structure of a Confluence space as a tree structure
Only structural metadata is returned (not content).
The response is akin to the sidebar in the Confluence UI.
Includes all pages, folders, whiteboards, databases,
smart links, etc. organized by parent-child relationships.
"""
client = ConfluenceClientV2(context.get_auth_token_or_empty())
space = await client.get_space(space_identifier)
tree = client.create_space_tree(space)
# Get root pages
root_pages = await client.get_root_pages_in_space(space["space"]["id"])
tree["children"] = client.convert_root_pages_to_tree_nodes(root_pages["pages"])
if not tree["children"]:
return {}
# Extract base URL for children URLs. The base URL is the space's URL.
root_page_url = tree["url"]
match = re.match(r"(.*?/spaces/[^/]+)", root_page_url)
children_base_url = match.group(1) if match else ""
# Get and descendants for each root page
await client.process_page_descendants(tree["children"], children_base_url)
return tree

View file

@ -0,0 +1,89 @@
import re
from arcade.sdk.errors import RetryableToolError, ToolExecutionError
def remove_none_values(data: dict) -> dict:
"""Remove all keys with None values from the dictionary."""
return {k: v for k, v in data.items() if v is not None}
def validate_ids(ids: list[str] | None, max_length: int) -> None:
"""Validate a list of IDs. The ids can be page ids, space ids, etc.
A valid id is a string that is a number.
Args:
ids: A list of IDs to validate.
Returns:
None
Raises:
ToolExecutionError: If any of the IDs are not valid.
RetryableToolError: If the number of IDs is greater than the max length.
"""
if not ids:
return
if len(ids) > max_length:
raise RetryableToolError(
message=f"The 'ids' parameter must have less than {max_length} items. Got {len(ids)}"
)
if any(not id_.isdigit() for id_ in ids):
raise ToolExecutionError(message="Invalid ID provided. IDs are numeric")
def build_child_url(base_url: str, child: dict) -> str | None:
"""Build URL for a child node based on its type and status.
Args:
base_url: The base URL for the Confluence space
child: A dictionary representing a Confluence content item
Returns:
The URL for the child, or None if it can't be determined
"""
if child["type"] in ("whiteboard", "database", "embed"):
return f"{base_url}/{child['type']}/{child['id']}"
elif child["type"] == "folder":
return None
elif child["type"] == "page":
parsed_title = re.sub(r"[ '\s]+", "+", child["title"].strip())
if child.get("status") == "draft":
return f"{base_url}/{child['type']}s/edit-v2/{child['id']}"
else:
return f"{base_url}/{child['type']}s/{child['id']}/{parsed_title}"
return None
def build_hierarchy(transformed_children: list, parent_id: str, parent_node: dict) -> None:
"""Build parent-child hierarchy from a flat list of descendants.
This function takes a flat list of items that have parent_id references and
builds a hierarchical tree structure. It modifies the parent_node in place.
Args:
transformed_children: List of child nodes with parent_id fields
parent_id: The ID of the parent node
parent_node: The parent node to attach direct children to
Returns:
None (modifies parent_node in place)
"""
# Create a map of children by their ID for efficient lookups
child_map = {child["id"]: child for child in transformed_children}
# Find all direct children of the given parent_id
direct_children = []
for child in transformed_children:
if child.get("parent_id") == parent_id:
direct_children.append(child)
elif child.get("parent_id") in child_map:
# Add child to its parent's children list
parent = child_map[child.get("parent_id")]
if "children" not in parent:
parent["children"] = []
parent["children"].append(child)
# Set the direct children on the parent node
parent_node["children"] = direct_children

View file

@ -0,0 +1,89 @@
# A conversation where a user asks for the structure of a space,
# and an LLM calls a tool to get the structure.
get_space_hierarchy_1 = [
{"role": "system", "content": "Today is 2025-05-16, Friday."},
{"role": "user", "content": "get structure of 98308"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_exyK4LmJEHSDn1Xw5oVfS9Xx",
"type": "function",
"function": {
"name": "Confluence_GetSpaceHierarchy",
"arguments": '{"space_identifier":"98308"}',
},
}
],
},
{
"role": "tool",
"content": '{"98418":{"children":[{"children":[],"id":"4653060","parent_id":"98418","parent_type":"TODO","title":"Mydb","type":"database","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/database/4653060"},{"children":[{"children":[],"id":"4685837","parent_id":"4653064","parent_type":"TODO","title":"10-03-98","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/4685837/10-03-98"}],"id":"4653064","parent_id":"98418","parent_type":"TODO","title":"DailyNotes","type":"folder","url":null},{"children":[{"children":[],"id":"5242883","parent_id":"5275650","parent_type":"TODO","title":"To be or not tobe","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242883/To+be+or+not+to+be"},{"children":[],"id":"5275653","parent_id":"5275650","parent_type":"TODO","title":"Two roadsdiverged","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/edit-v2/5275653"},{"children":[],"id":"5242903","parent_id":"5275650","parent_type":"TODO","title":"Eternity","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242903/Eternity"},{"children":[],"id":"5242913","parent_id":"5275650","parent_type":"TODO","title":"BlackHole","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242913/Black+Hole"}],"id":"5275650","parent_id":"98418","parent_type":"TODO","title":"Poetry","type":"folder","url":null},{"children":[],"id":"5275676","parent_id":"98418","parent_type":"TODO","title":"Drawings","type":"whiteboard","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/whiteboard/5275676"}],"id":"98418","parent_id":null,"parent_type":null,"title":"testspace Home","type":"page","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/overview"}}', # noqa: E501
"tool_call_id": "call_exyK4LmJEHSDn1Xw5oVfS9Xx",
"name": "Confluence_GetSpaceHierarchy",
},
{
"role": "assistant",
"content": 'Here is the structure for the space with the identifier 98308:\n\n- **testspace Home** [Page](https://ericconfluence.atlassian.net/wiki/spaces/testspace/overview)\n - **My db** [Database](https://ericconfluence.atlassian.net/wiki/spaces/testspace/database/4653060)\n - **Daily Notes** (Folder)\n - [10-03-98](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/4685837/10-03-98) (Page)\n - **Poetry** (Folder)\n - [To be or not to be](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242883/To+be+or+not+to+be) (Page)\n - [Two roads diverged](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/edit-v2/5275653) (Page)\n - [Eternity](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242903/Eternity) (Page)\n - [Black Hole](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5242913/Black+Hole) (Page)\n - **Drawings** [Whiteboard](https://ericconfluence.atlassian.net/wiki/spaces/testspace/whiteboard/5275676)\n\nThis structure includes various elements such as pages, a database, folders, and a whiteboard within the "testspace" space.', # noqa: E501
},
]
# A conversation where a user asks the assistant to create a page with content
create_page_1 = [
{
"role": "user",
"content": "make a new page within the testspace space under the 4653064 folder. Within it, make a list of all the TODOs for the day which are 1. Write a blog 2. Walk the dog", # noqa: E501
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_xuvOuXpLAmX57RO1zj5hqQ1r",
"type": "function",
"function": {
"name": "Confluence_CreatePage",
"arguments": '{"space_identifier":"testspace","title":"TODOs forToday","content":"<h1>TODOs for Today</h1><ol><li>Write a blog</li><li>Walk thedog</li></ol>","parent_id":"4653064"}', # noqa: E501
},
}
],
},
{
"role": "tool",
"content": '{"page":{"authorId":"712020:93b309a8-e9af-4615-b289-07bf23e6f13f","body":{"storage":{"representation":"storage","value":"\\u003ch1\\u003eTODOs for Today\\u003c/h1\\u003e\\u003col\\u003e\\u003cli\\u003eWrite a blog\\u003c/li\\u003e\\u003cli\\u003eWalk the dog\\u003c/li\\u003e\\u003c/ol\\u003e"}},"createdAt":"2025-05-16T22:29:59.969Z","id":"5439489","lastOwnerId":null,"ownerId":"712020:93b309a8-e9af-4615-b289-07bf23e6f13f","parentId":"4653064","parentType":"folder","position":1675,"spaceId":"98308","status":"current","title":"TODOs for Today","url":"https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5439489/TODOs+for+Today","version":{"authorId":"712020:93b309a8-e9af-4615-b289-07bf23e6f13f","createdAt":"2025-05-16T22:29:59.969Z","message":"","minorEdit":false,"ncsStepVersion":"1","number":1}}}', # noqa: E501
"tool_call_id": "call_xuvOuXpLAmX57RO1zj5hqQ1r",
"name": "Confluence_CreatePage",
},
{
"role": "assistant",
"content": 'I have created a new page titled "TODOs for Today" in the "testspace" space, under the folder with ID 4653064. The list of your tasks for the day is included:\n\n1. Write a blog\n2. Walk the dog\n\nYou can view the page [here](https://ericconfluence.atlassian.net/wiki/spaces/testspace/pages/5439489/TODOs+for+Today).', # noqa: E501
},
]
# A conversation where a user asks the assistant to list one space.
list_spaces_1 = [
{"role": "system", "content": "Today is 2025-05-19, Monday."},
{"role": "user", "content": "get 1 space"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_ur97pVeXzdB7ECNrr0zQBJmE",
"type": "function",
"function": {"name": "Confluence_ListSpaces", "arguments": '{"limit":1}'},
}
],
},
{
"role": "tool",
"content": '{"pagination_token":"eyJpZCI6MjMyNjc5NCwic3BhY2VTb3J0T3JkZXIiOnsiZmllbGQiOiJOQU1FIiwiZGlyZWN0aW9uIjoiQVNDRU5ESU5HIn0sInNwYWNlU29ydE9yZGVyVmFsdWUiOiJlcmljYXJjYWRlIn0=","spaces":[{"authorId":"712020:f889c42e-f53e-4500-a7c6-706c3ef10951","createdAt":"2025-05-12T21:52:24.230Z","currentActiveAlias":"~712020f889c42ef53e4500a7c6706c3ef10951","description":null,"homepageId":"2327032","icon":null,"id":"2326794","key":"~712020f889c42ef53e4500a7c6706c3ef10951","name":"ericarcade","spaceOwnerId":"712020:f889c42e-f53e-4500-a7c6-706c3ef10951","status":"current","type":"personal","url":"https://ericconfluence.atlassian.net/wiki/spaces/~712020f889c42ef53e4500a7c6706c3ef10951"}]}',
"tool_call_id": "call_ur97pVeXzdB7ECNrr0zQBJmE",
"name": "Confluence_ListSpaces",
},
{
"role": "assistant",
"content": "Here is one space from Confluence:\n\n- **Name**: ericarcade\n- **Key**: ~712020f889c42ef53e4500a7c6706c3ef10951\n- **Type**:Personal\n- **Status**: Current\n- **Author ID**: 712020:f889c42e-f53e-4500-a7c6-706c3ef10951\n- **Created At**: 2025-05-12\n- **Homepage ID**:2327032\n- **URL**: [ericarcade Space](https://ericconfluence.atlassian.net/wiki/spaces/~712020f889c42ef53e4500a7c6706c3ef10951)", # noqa: E501
},
]

View file

@ -0,0 +1,36 @@
from dataclasses import dataclass
from typing import Any
from arcade.sdk.eval.critic import Critic
@dataclass
class ListCritic(Critic):
"""
A critic for comparing two lists.
"""
def __init__(
self,
critic_field: str,
weight: float = 1.0,
order_matters: bool = True,
duplicates_matter: bool = True,
case_sensitive: bool = False,
):
self.critic_field = critic_field
self.weight = weight
self.order_matters = order_matters
self.duplicates_matter = duplicates_matter
self.case_sensitive = case_sensitive
def evaluate(self, expected: list[Any], actual: list[Any]) -> dict[str, float | bool]:
if not self.case_sensitive:
actual = [item.lower() for item in actual]
expected = [item.lower() for item in expected]
match = actual == expected if self.order_matters else set(actual) == set(expected)
if self.duplicates_matter:
match = match and len(actual) == len(expected)
return {"match": match, "score": self.weight if match else 0.0}

View file

@ -0,0 +1,207 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade.sdk.eval.critic import BinaryCritic, SimilarityCritic
import arcade_confluence
from arcade_confluence.enums import PageUpdateMode
from arcade_confluence.tools import (
create_page,
get_page,
get_pages_by_id,
rename_page,
update_page_content,
)
from evals.conversations import create_page_1, get_space_hierarchy_1
from evals.critics import ListCritic
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_confluence)
@tool_eval()
def confluence_get_page_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence get_page tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get page by ID",
user_message="Get page 65816",
expected_tool_calls=[ExpectedToolCall(func=get_page, args={"page_identifier": 65816})],
rubric=rubric,
critics=[BinaryCritic(critic_field="page_identifier", weight=1)],
)
suite.add_case(
name="Get page by title",
user_message="Get my 'Poem - May 24th' page",
expected_tool_calls=[
ExpectedToolCall(func=get_page, args={"page_identifier": "Poem - May 24th"})
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="page_identifier", weight=1),
],
)
suite.add_case(
name="Get page based on previous conversation",
user_message=(
"Get the content of my daily note. You MUST use the page's title when getting the page."
),
expected_tool_calls=[ExpectedToolCall(func=get_page, args={"page_identifier": "10-03-98"})],
rubric=rubric,
critics=[
BinaryCritic(critic_field="page_identifier", weight=1),
],
additional_messages=get_space_hierarchy_1,
)
return suite
@tool_eval()
def confluence_get_multiple_pages_by_id_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence get_multiple_pages_by_id tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get multiple pages by ID",
user_message="Get 98418, 4685837, 5242883, 5275653, 5242903, 5242913 pages",
expected_tool_calls=[
ExpectedToolCall(
func=get_pages_by_id,
args={"page_ids": ["98418", "4685837", "5242883", "5275653", "5242903", "5242913"]},
)
],
rubric=rubric,
critics=[
ListCritic(critic_field="page_ids", order_matters=False, weight=1),
],
)
suite.add_case(
name="Get multiple pages by ID with existing conversation",
user_message=("Get the content of all pages in the space except 4685837"),
expected_tool_calls=[
ExpectedToolCall(
func=get_pages_by_id,
args={"page_ids": ["98418", "5242883", "5275653", "5242903", "5242913"]},
)
],
rubric=rubric,
critics=[
ListCritic(critic_field="page_ids", order_matters=False, weight=1),
],
additional_messages=get_space_hierarchy_1,
)
return suite
@tool_eval()
def confluence_create_page_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence create_page tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Create a page",
user_message=(
"Make a page within the softwareeng space called 'TODOs' under the 4830960 folder. "
"Within it, make a list of all the TODOs for the day which are "
"1. Write a blog post about the future of AI"
"2. Write an agent that calls my mom at 5:30 every Friday evening"
),
expected_tool_calls=[
ExpectedToolCall(
func=create_page,
args={
"space_identifier": "softwareeng",
"title": "TODOs",
"content": "1. Write a blog post about the future of AI\n2. Write an agent that calls my mom at 5:30 every Friday evening", # noqa: E501
"parent_id": "4830960",
},
)
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="space_identifier", weight=1 / 4),
BinaryCritic(critic_field="title", weight=1 / 4),
SimilarityCritic(critic_field="content", weight=1 / 4),
BinaryCritic(critic_field="parent_id", weight=1 / 4),
],
)
return suite
@tool_eval()
def confluence_update_page_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence update page tools evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Update page content",
user_message="Thanks, now append '3. Walk the dog' to the end of the list",
expected_tool_calls=[
ExpectedToolCall(
func=update_page_content,
args={
"page_identifier": "5439489",
"content": "3. Walk the dog",
"update_mode": PageUpdateMode.APPEND,
},
)
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="page_identifier", weight=1 / 3),
SimilarityCritic(critic_field="content", weight=1 / 3),
BinaryCritic(critic_field="update_mode", weight=1 / 3),
],
additional_messages=create_page_1,
)
suite.extend_case(
name="Rename page",
user_message="Actually, rename it to 'My TODOs'",
expected_tool_calls=[
ExpectedToolCall(
func=rename_page,
args={"page_identifier": "5439489", "title": "My TODOs"},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="page_identifier", weight=1 / 2),
BinaryCritic(critic_field="title", weight=1 / 2),
],
)
return suite

View file

@ -0,0 +1,111 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade.sdk.eval.critic import BinaryCritic
import arcade_confluence
from arcade_confluence.tools import (
search_content,
)
from evals.critics import ListCritic
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_confluence)
@tool_eval()
def confluence_search_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence search content tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Search for content - easy case",
user_message="Find all pages that contain 'Arcade.dev'",
expected_tool_calls=[
ExpectedToolCall(func=search_content, args={"can_contain_any": ["Arcade.dev"]})
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="can_contain_any", weight=1),
],
)
suite.add_case(
name="Search for content - medium case",
user_message=("Find 20 pages that contain 'Arcade' or 'AI', or 'tool calls'"),
expected_tool_calls=[
ExpectedToolCall(
func=search_content,
args={
"can_contain_any": ["Arcade", "AI", "tool calls"],
"limit": 20,
},
)
],
rubric=rubric,
critics=[
ListCritic(
critic_field="can_contain_any",
weight=0.9,
case_sensitive=False,
order_matters=False,
),
BinaryCritic(critic_field="limit", weight=0.1),
],
)
suite.add_case(
name="Search for content - hard case",
user_message=(
"Look for 25 databases that have 'How to', "
"and also have 'carborator' in the content and "
"also has one of the following: 'money', 'redbull gives you wings', "
"'honey hole', 'don't snap the pasta!'."
),
expected_tool_calls=[
ExpectedToolCall(
func=search_content,
args={
"must_contain_all": ["carborator", "How to"],
"can_contain_any": [
"money",
"redbull gives you wings",
"honey hole",
"don't snap the pasta!",
],
},
),
],
rubric=rubric,
critics=[
ListCritic(
critic_field="must_contain_all",
weight=0.5,
case_sensitive=False,
order_matters=False,
),
ListCritic(
critic_field="can_contain_any",
weight=0.5,
case_sensitive=False,
order_matters=False,
),
],
)
return suite

View file

@ -0,0 +1,113 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade.sdk.eval.critic import BinaryCritic
import arcade_confluence
from arcade_confluence.tools import (
get_space,
get_space_hierarchy,
list_spaces,
)
from evals.conversations import list_spaces_1
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_confluence)
@tool_eval()
def confluence_get_space_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence get_space tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get two spaces - one by ID, and one by key",
user_message="Get spaces 3498573 and 'Poetry'",
expected_tool_calls=[
ExpectedToolCall(func=get_space, args={"space_identifier": 3498573}),
ExpectedToolCall(func=get_space, args={"space_identifier": "Poetry"}),
],
rubric=rubric,
critics=[BinaryCritic(critic_field="space_identifier", weight=1)],
)
return suite
@tool_eval()
def confluence_list_spaces_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence list_spaces tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="List the next space using a pagination token",
user_message="get the next one",
expected_tool_calls=[
ExpectedToolCall(
func=list_spaces,
args={
"limit": 1,
"pagination_token": "eyJpZCI6MjMyNjc5NCwic3BhY2VTb3J0T3JkZXIiOnsiZmllbGQiOiJOQU1FIiwiZGlyZWN0aW9uIjoiQVNDRU5ESU5HIn0sInNwYWNlU29ydE9yZGVyVmFsdWUiOiJlcmljYXJjYWRlIn0=", # noqa: E501
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="limit", weight=0.3),
BinaryCritic(critic_field="pagination_token", weight=0.7),
],
additional_messages=list_spaces_1,
)
return suite
@tool_eval()
def confluence_get_space_hierarchy_eval_suite() -> EvalSuite:
suite = EvalSuite(
name="Confluence get_space_hierarchy tool evaluation",
system_message="You are an AI assistant with access to Confluence tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get the hierarchy of a space",
user_message=(
"What is the best file location within my 'Poetry' space to create a new page "
"named 'Rough Draft - Poem for King Henry VIII'?"
),
expected_tool_calls=[
ExpectedToolCall(
func=get_space_hierarchy,
args={
"space_identifier": "Poetry",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="space_identifier", weight=1),
],
)
return suite

View file

@ -0,0 +1,41 @@
[tool.poetry]
name = "arcade_confluence"
version = "0.0.1"
description = "Arcade.dev LLM tools for Confluence"
authors = ["Arcade <dev@arcade.dev>"]
[tool.poetry.dependencies]
python = "^3.10"
arcade-ai = "^1.0.5"
httpx = "^0.27.2"
[tool.poetry.dev-dependencies]
pytest = "^8.3.0"
pytest-cov = "^4.0.0"
mypy = "^1.5.1"
pre-commit = "^3.4.0"
tox = "^4.11.1"
ruff = "^0.7.4"
pytest-asyncio = "^0.24.0"
[build-system]
requires = ["poetry-core>=1.0.0,<2.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.mypy]
files = ["arcade_confluence/**/*.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

View file

View file

@ -0,0 +1,120 @@
from unittest.mock import patch
import pytest
from arcade.sdk.errors import ToolExecutionError
from arcade_confluence.client import ConfluenceClientV1
@pytest.fixture
def client_v1() -> ConfluenceClientV1:
"""Fixture that provides a ConfluenceClientV1 instance with mocked cloud_id."""
with patch("arcade_confluence.client.ConfluenceClient._get_cloud_id", return_value=None):
return ConfluenceClientV1("fake-token")
@pytest.mark.parametrize(
"query, enable_fuzzy, expected_cql",
[
("foo", False, '(text ~ "foo" OR title ~ "foo" OR space.title ~ "foo")'),
("foo bar", False, '(text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")'),
("foo", True, '(text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~")'),
("foo bar", True, '(text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")'),
],
)
def test_build_query_cql(client_v1: ConfluenceClientV1, query, enable_fuzzy, expected_cql) -> None:
cql = client_v1._build_query_cql(query, enable_fuzzy)
assert cql == expected_cql
@pytest.mark.parametrize(
"queries, enable_fuzzy, expected_cql",
[
(
["foo", "foo bar"],
False,
'((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
["foo", "foo bar"],
True,
'((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
],
)
def test_build_and_cql(client_v1: ConfluenceClientV1, queries, enable_fuzzy, expected_cql) -> None:
cql = client_v1._build_and_cql(queries, enable_fuzzy)
assert cql == expected_cql
@pytest.mark.parametrize(
"queries, enable_fuzzy, expected_cql",
[
(
["foo", "foo bar"],
False,
'((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
["foo", "foo bar"],
True,
'((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
],
)
def test_build_or_cql(client_v1: ConfluenceClientV1, queries, enable_fuzzy, expected_cql) -> None:
cql = client_v1._build_or_cql(queries, enable_fuzzy)
assert cql == expected_cql
@pytest.mark.parametrize(
"must_contain_all, can_contain_any, enable_fuzzy, expected_cql",
[
(None, None, False, ""),
(
["foo", "foo bar"],
None,
False,
'((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
["foo", "foo bar"],
None,
True,
'((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
None,
["foo", "foo bar"],
False,
'((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
None,
["foo", "foo bar"],
True,
'((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar"))', # noqa: E501
),
(
["foo", "foo bar"],
["foo", "foo bar"],
False,
'(((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")) AND ((text ~ "foo" OR title ~ "foo" OR space.title ~ "foo") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")))', # noqa: E501
),
(
["foo", "foo bar"],
["foo", "foo bar"],
True,
'(((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") AND (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")) AND ((text ~ "foo~" OR title ~ "foo~" OR space.title ~ "foo~") OR (text ~ "foo bar" OR title ~ "foo bar" OR space.title ~ "foo bar")))', # noqa: E501
),
],
)
def test_construct_cql(
client_v1: ConfluenceClientV1, must_contain_all, can_contain_any, enable_fuzzy, expected_cql
) -> None:
if not expected_cql:
with pytest.raises(ToolExecutionError):
client_v1.construct_cql(must_contain_all, can_contain_any, enable_fuzzy)
else:
cql = client_v1.construct_cql(must_contain_all, can_contain_any, enable_fuzzy)
assert cql == expected_cql

View file

@ -0,0 +1,103 @@
from unittest.mock import patch
import pytest
from arcade_confluence.client import ConfluenceClientV2
@pytest.mark.asyncio
@pytest.mark.parametrize(
"space_identifier, is_id",
[
(
"12345",
True,
),
(
"test-space",
False,
),
],
)
async def test_get_space_id(space_identifier, is_id) -> None:
with patch("arcade_confluence.client.ConfluenceClient._get_cloud_id", return_value=None):
client_v2 = ConfluenceClientV2("fake-token")
with patch(
"arcade_confluence.client.ConfluenceClientV2.get_space_by_key",
return_value={"space": {"id": "12345"}},
):
space_id = await client_v2.get_space_id(space_identifier)
if is_id:
assert space_id == space_identifier
else:
assert space_id == "12345"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"page_identifier, is_id",
[
(
"67890",
True,
),
(
"test-page",
False,
),
],
)
async def test_get_page_id(page_identifier, is_id) -> None:
with patch("arcade_confluence.client.ConfluenceClient._get_cloud_id", return_value=None):
client_v2 = ConfluenceClientV2("fake-token")
with patch(
"arcade_confluence.client.ConfluenceClientV2.get_page_by_title",
return_value={"page": {"id": "67890"}},
):
page_id = await client_v2.get_page_id(page_identifier)
if is_id:
assert page_id == page_identifier
else:
assert page_id == "67890"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"space_identifier, is_id",
[
(
"12345",
True,
),
(
"test-space",
False,
),
],
)
async def test_get_space(space_identifier, is_id) -> None:
with patch("arcade_confluence.client.ConfluenceClient._get_cloud_id", return_value=None):
client_v2 = ConfluenceClientV2("fake-token")
mock_space = {"space": {"id": "12345", "key": "TEST"}}
with (
patch(
"arcade_confluence.client.ConfluenceClientV2.get_space_by_id",
return_value=mock_space,
) as mock_by_id,
patch(
"arcade_confluence.client.ConfluenceClientV2.get_space_by_key",
return_value=mock_space,
) as mock_by_key,
):
result = await client_v2.get_space(space_identifier)
if is_id:
mock_by_id.assert_called_once_with(space_identifier)
mock_by_key.assert_not_called()
else:
mock_by_id.assert_not_called()
mock_by_key.assert_called_once_with(space_identifier)
assert result == mock_space

View file

@ -0,0 +1,195 @@
import pytest
from arcade.sdk.errors import RetryableToolError, ToolExecutionError
from arcade_confluence.utils import build_child_url, build_hierarchy, validate_ids
@pytest.mark.parametrize(
"ids, max_length, expected_error",
[
(None, 250, None),
(["123", "456"], 250, None),
(["123", "foo"], 250, ToolExecutionError),
(["123", "456"], 1, RetryableToolError),
],
)
def test_validate_ids(ids: list[str], max_length: int, expected_error: Exception | None) -> None:
if expected_error:
with pytest.raises(expected_error):
validate_ids(ids, max_length)
else:
validate_ids(ids, max_length)
@pytest.mark.parametrize(
"base_url, child, expected",
[
( # Published page
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "page", "title": "Test Title-1", "id": "123", "status": "current"},
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT/pages/123/Test+Title-1",
),
( # Draft page
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "page", "title": "Test Title-1", "id": "123", "status": "draft"},
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT/pages/edit-v2/123",
),
( # Whiteboard
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "whiteboard", "title": "Test Title-1", "id": "123", "status": "current"},
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT/whiteboard/123",
),
( # Database
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "database", "title": "Test Title-1", "id": "123", "status": "current"},
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT/database/123",
),
( # Embed
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "embed", "title": "Test Title-1", "id": "123", "status": "current"},
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT/embed/123",
),
( # Folder
"https://tes.atlassian.net/wiki/spaces/SOFTWAREDEVELOPMENT",
{"type": "folder", "title": "Test Title-1", "id": "123", "status": "current"},
None, # Folders don't have URLs
),
],
)
def test_build_child_url(base_url: str, child: dict, expected: str) -> None:
assert build_child_url(base_url, child) == expected
@pytest.mark.parametrize(
"input_transformed_children, input_parent_id, input_parent_node, expected_parent_node",
[
( # Parent node is a leaf
[],
"2195457",
{
"title": "A One Sentence Story About Trees",
"id": "2195457",
"type": "page",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2195457/A+One+Sentence+Story+About+Trees",
"children": [],
},
{
"title": "A One Sentence Story About Trees",
"id": "2195457",
"type": "page",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2195457/A+One+Sentence+Story+About+Trees",
"children": [],
},
),
(
[
{
"title": "The Importance of Trees",
"id": "2555906",
"type": "page",
"parent_id": "2162740",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2555906/The+Importance+of+Trees",
"children": [],
"status": "current",
},
{
"title": "Erics page",
"id": "2686977",
"type": "page",
"parent_id": "2555906",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2686977/Erics+page",
"children": [],
"status": "current",
},
{
"title": "Erics page",
"id": "2719745",
"type": "page",
"parent_id": "2555906",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/edit-v2/2719745",
"children": [],
"status": "draft",
},
{
"title": "Execute tools",
"id": "2621441",
"type": "page",
"parent_id": "2162740",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2621441/Execute+tools",
"children": [],
"status": "current",
},
],
"2162740",
{
"title": "Trees",
"id": "2162740",
"type": "page",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2162740/Trees",
"children": [],
},
{
"title": "Trees",
"id": "2162740",
"type": "page",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2162740/Trees",
"children": [
{
"title": "The Importance of Trees",
"id": "2555906",
"type": "page",
"parent_id": "2162740",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2555906/The+Importance+of+Trees",
"children": [
{
"title": "Erics page",
"id": "2686977",
"type": "page",
"parent_id": "2555906",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2686977/Erics+page",
"children": [],
"status": "current",
},
{
"title": "Erics page",
"id": "2719745",
"type": "page",
"parent_id": "2555906",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/edit-v2/2719745",
"children": [],
"status": "draft",
},
],
"status": "current",
},
{
"title": "Execute tools",
"id": "2621441",
"type": "page",
"parent_id": "2162740",
"parent_type": "TODO",
"url": "https://ericconfluence.atlassian.net/wiki/spaces/SOFTWAREDE/pages/2621441/Execute+tools",
"children": [],
"status": "current",
},
],
},
),
],
)
def test_build_hierarchy(
input_transformed_children: list[dict],
input_parent_id: int,
input_parent_node: dict,
expected_parent_node: dict,
) -> None:
# build_hierarchy modifies the input_parent_node in-place
build_hierarchy(input_transformed_children, input_parent_id, input_parent_node)
assert input_parent_node == expected_parent_node