"""
title: Confluence page
description: This tool allows you to retrieve content from a specific page on Confluence.
repository: https://github.com/RomainNeup/open-webui-utilities
author: @romainneup
author_url: https://github.com/RomainNeup
funding_url: https://github.com/sponsors/RomainNeup
requirements: markdownify
version: 0.1.1
changelog:
- 0.0.1 - Initial code base.
- 0.0.2 - Fix Valves variables
- 0.1.0 - Split Confuence search and Confluence get page
- 0.1.1 - Add support for Personal Access Token authentication and user settings
"""
import base64
import json
import requests
from typing import Awaitable, Callable, Dict, Any
from pydantic import BaseModel, Field
from markdownify import markdownify
class EventEmitter:
def __init__(self, event_emitter: Callable[[dict], Awaitable[None]]):
self.event_emitter = event_emitter
pass
async def emit_status(self, description: str, done: bool, error: bool = False):
await self.event_emitter(
{
"data": {
"description": f"{done and (error and '❌' or '✅') or '🔎'} {description}",
"status": done and "complete" or "in_progress",
"done": done,
},
"type": "status",
}
)
async def emit_message(self, content: str):
await self.event_emitter({"data": {"content": content}, "type": "message"})
async def emit_source(self, name: str, url: str, content: str, html: bool = False):
await self.event_emitter(
{
"type": "citation",
"data": {
"document": [content],
"metadata": [{"source": url, "html": html}],
"source": {"name": name},
},
}
)
class Confluence:
def __init__(
self, username: str, api_key: str, base_url: str, api_key_auth: bool = True
):
self.base_url = base_url
self.headers = self.authenticate(username, api_key, api_key_auth)
pass
def get(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
url = f"{self.base_url}/rest/api/{endpoint}"
response = requests.get(url, params=params, headers=self.headers)
if not response.ok:
raise Exception(f"Failed to get data from Confluence: {response.text}")
return response.json()
def get_page(self, page_id: str) -> Dict[str, Any]:
endpoint = f"content/{page_id}"
params = {"expand": "body.view", "include-version": "false"}
result = self.get(endpoint, params)
return {
"id": result["id"],
"title": result["title"],
"body": markdownify(result["body"]["view"]["value"]),
"link": f"{self.base_url}{result['_links']['webui']}",
}
def authenticate_api_key(self, username: str, api_key: str) -> Dict[str, str]:
auth_string = f"{username}:{api_key}"
encoded_auth_string = base64.b64encode(auth_string.encode("utf-8")).decode(
"utf-8"
)
return {"Authorization": "Basic " + encoded_auth_string}
def authenticate_personal_access_token(self, access_token: str) -> Dict[str, str]:
return {"Authorization": f"Bearer {access_token}"}
def authenticate(self, username: str, api_key: str, api_key_auth: bool) -> Dict[str, str]:
if api_key_auth:
return self.authenticate_api_key(username, api_key)
else:
return self.authenticate_personal_access_token(api_key)
class Tools:
def __init__(self):
self.valves = self.Valves()
pass
class Valves(BaseModel):
base_url: str = Field(
"https://example.atlassian.net/wiki",
description="The base URL of your Confluence instance",
)
username: str = Field(
"
[email protected]",
description="Default username (leave empty for personal access token)",
)
api_key: str = Field(
"ABCD1234", description="Default API key or personal access token"
)
pass
class UserValves(BaseModel):
api_key_auth: bool = Field(
True,
description="Use API key authentication; disable this to use a personal access token instead.",
)
username: str = Field(
"",
description="Username, typically your email address; leave empty if using a personal access token or default settings.",
)
api_key: str = Field(
"",
description="API key or personal access token; leave empty to use the default settings.",
)
pass
# Get content from Confluence
async def get_confluence_page(
self,
page_id: str,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {}
) -> str:
"""
Get the content of a page on Confluence. This returns the content of a specific page on Confluence.
Use it to get the content of a specific page on Confluence. When a user requests a specific page, this must be used.
:param page_id: The ID of the page on Confluence
:return: The content of the page on Confluence in JSON format (title, body, link). If the page is not found, an error message is returned.
"""
event_emitter = EventEmitter(__event_emitter__)
# Get the username and API key
if __user__ and "valves" in __user__:
user_valves = __user__["valves"]
api_key_auth = user_valves.api_key_auth
api_username = user_valves.username or self.valves.username
api_key = user_valves.api_key or self.valves.api_key
else:
api_username = self.valves.username
api_key = self.valves.api_key
api_key_auth = True
if (api_key_auth and not api_username) or not api_key:
await event_emitter.emit_status(
"Please provide a username and API key or personal access token.", True, True
)
return "Error: Please provide a username and API key or personal access token."
confluence = Confluence(
api_username, api_key, self.valves.base_url, api_key_auth
)
await event_emitter.emit_status(f"Retrieving page '{page_id}' from Confluence...", False)
try:
result = confluence.get_page(page_id)
await event_emitter.emit_status(f"Retrieved page '{page_id}' from Confluence.", True)
await event_emitter.emit_source(result["title"], result["link"], result["body"])
return json.dumps(result)
except Exception as e:
await event_emitter.emit_status(f"Failed to retrieve page '{page_id}': {e}.", True, True)
return f"Error: {e}"
