"""
title: Jira Search
description: This tool allows you to search issues from 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 - Implement Jira search
- 0.1.0 - Add support for Personal Access Token authentication and user settings
- 0.1.1 - Limit setting for search results
"""
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)
return response.json()
def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
endpoint = "search"
terms = query.split()
if terms:
cql_terms = " OR ".join([f'text ~ "{term}"' for term in terms])
else:
cql_terms = f'text ~ "{query}"'
params = {"jql": f"{cql_terms}", "maxResults": limit}
rawResponse = self.get(endpoint, params)
response = []
for item in rawResponse["issues"]:
response.append(item["key"])
return response
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"
)
result_limit: int = Field(
5, description="The maximum number of search results to return", required=True
)
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 search_jira(
self,
query: str,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> str:
"""
Search for a query on Jira. This returns the result of the search on Jira.
Use it to search for a query on Jira. When a user mentions a search on Jira, this must be used.
Note: This returns a list of issues that match the search query.
:param query: The text to search for on Jira. MUST be a string.
:return: A list of search results from Jira in JSON format (key). If no results are found, an empty list 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
jira = Jira(api_username, api_key, self.valves.base_url, api_key_auth)
await event_emitter.emit_status(
f"Searching for '{query}' on Jira...", False
)
try:
searchResponse = jira.search(query, self.valves.result_limit)
await event_emitter.emit_status(
f"Search for '{query}' on Jira complete. ({len(searchResponse)} results found)",
True,
)
return json.dumps(searchResponse)
except Exception as e:
await event_emitter.emit_status(
f"Failed to search for '{query}' on Jira: {e}", True, True
)
return f"Error: {e}"
