""" title: Jira Issue description: This tool allows you to get information about a specific issue on Jira. repository: https://github.com/RomainNeup/open-webui-utilities author: @romainneup author_url: https://github.com/RomainNeup funding_url: https://github.com/sponsors/RomainNeup version: 0.1.0 changelog: - 0.0.1 - Initial code base. - 0.0.2 - Split Jira search and Jira get issue - 0.1.0 - Add support for Personal Access Token authentication and user settings """ import base64 import json from typing import Any, Awaitable, Callable, Dict import requests from pydantic import BaseModel, Field 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 Jira: 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]): url = f"{self.base_url}/rest/api/3/{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_issue(self, issue_id: str): endpoint = f"issue/{issue_id}" result = self.get( endpoint, {"fields": "summary,description,status", "expand": "renderedFields"}, ) return { "title": result["fields"]["summary"], "description": result["renderedFields"]["description"], "status": result["fields"]["status"]["name"], "link": f"{self.base_url}/browse/{issue_id}", } 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( "example@example.com", 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 async def get_issue( self, issue_id: str, __event_emitter__: Callable[[dict], Awaitable[None]], __user__: dict = {}, ): """ Get a Jira issue by its ID. The response includes the title, description as HTML, status, and link to the issue. :param issue_id: The ID of the issue. :return: A response in JSON format (title, description, status, link). """ 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 jira = Jira(api_username, api_key, self.valves.base_url, api_key_auth) await event_emitter.emit_status(f"Getting issue {issue_id}", False) try: response = jira.get_issue(issue_id) await event_emitter.emit_status(f"Got issue {issue_id}", True) await event_emitter.emit_source( response["title"], response["link"], response["description"], True ) return json.dumps(response) except Exception as e: await event_emitter.emit_status( f"Failed to get issue {issue_id}: {e}", True, True ) return f"Error: {e}"