Merge branch 'main' into fix/ollama-model-copy-for-playground

cookbook/tools/confluence_tools.py

@@ -0,0 +1,22 @@
+from phi.agent import Agent
+from phi.tools.confluence import ConfluenceTools
+
+
+agent = Agent(
+    name="Confluence agent",
+    tools=[ConfluenceTools()],
+    show_tool_calls=True,
+    markdown=True,
+)
+
+## getting space details
+agent.print_response("How many spaces are there and what are their names?")
+
+## getting page_content
+agent.print_response("What is the content present in page 'Large language model in LLM space'")
+
+## getting page details in a particular space
+agent.print_response("Can you extract all the page names from 'LLM' space")
+
+## creating a new page in a space
+agent.print_response("Can you create a new page named 'TESTING' in 'LLM' space")

phi/tools/confluence.py

@@ -0,0 +1,174 @@
+from phi.tools import Toolkit
+from phi.utils.log import logger
+from typing import Optional
+from os import getenv
+import json
+
+try:
+    from atlassian import Confluence
+except (ModuleNotFoundError, ImportError):
+    raise ImportError("atlassian-python-api not install . Please install using `pip install atlassian-python-api`")
+
+
+class ConfluenceTools(Toolkit):
+    def __init__(
+        self,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        url: Optional[str] = None,
+        api_key: Optional[str] = None,
+    ):
+        """Initialize Confluence Tools with authentication credentials.
+
+        Args:
+            username (str, optional): Confluence username. Defaults to None.
+            password (str, optional): Confluence password. Defaults to None.
+            url (str, optional): Confluence instance URL. Defaults to None.
+            api_key (str, optional): Confluence API key. Defaults to None.
+
+        Notes:
+            Credentials can be provided either through method arguments or environment variables:
+            - CONFLUENCE_URL
+            - CONFLUENCE_USERNAME
+            - CONFLUENCE_API_KEY
+        """
+
+        super().__init__(name="confluence_tools")
+        self.url = url or getenv("CONFLUENCE_URL")
+        self.username = username or getenv("CONFLUENCE_USERNAME")
+        self.password = api_key or getenv("CONFLUENCE_API_KEY") or password or getenv("CONFLUENCE_PASSWORD")
+
+        if not self.url:
+            logger.error(
+                "Confluence URL not provided. Pass it in the constructor or set CONFLUENCE_URL in environment variable"
+            )
+
+        if not self.username:
+            logger.error(
+                "Confluence username not provided. Pass it in the constructor or set CONFLUENCE_USERNAME in environment variable"
+            )
+
+        if not self.password:
+            logger.error("Confluence API KEY or password not provided")
+
+        self.confluence = Confluence(url=self.url, username=self.username, password=self.password)
+
+        self.register(self.get_page_content)
+        self.register(self.get_space_key)
+        self.register(self.create_page)
+        self.register(self.update_page)
+        self.register(self.get_all_space_detail)
+        self.register(self.get_all_page_from_space)
+
+    def get_page_content(self, space_name: str, page_title: str, expand: Optional[str] = "body.storage"):
+        """Retrieve the content of a specific page in a Confluence space.
+
+        Args:
+            space_name (str): Name of the Confluence space.
+            page_title (str): Title of the page to retrieve.
+            expand (str, optional): Fields to expand in the page response. Defaults to "body.storage".
+
+        Returns:
+            str: JSON-encoded page content or error message.
+        """
+        try:
+            logger.info(f"Retrieving page content from space '{space_name}'")
+            key = self.get_space_key(space_name=space_name)
+            page = self.confluence.get_page_by_title(key, page_title, expand=expand)
+            if page:
+                logger.info(f"Successfully retrieved page '{page_title}' from space '{space_name}'")
+                return json.dumps(page)
+
+            logger.warning(f"Page '{page_title}' not found in space '{space_name}'")
+            return json.dumps({"error": f"Page '{page_title}' not found in space '{space_name}'"})
+
+        except Exception as e:
+            logger.error(f"Error retrieving page '{page_title}': {e}")
+            return json.dumps({"error": str(e)})
+
+    def get_all_space_detail(self):
+        """Retrieve details about all Confluence spaces.
+
+        Returns:
+            str: List of space details as a string.
+        """
+        logger.info("Retrieving details for all Confluence spaces")
+        results = self.confluence.get_all_spaces()["results"]
+        return str(results)
+
+    def get_space_key(self, space_name: str):
+        """Get the space key for a particular Confluence space.
+
+        Args:
+            space_name (str): Name of the space whose key is required.
+
+        Returns:
+            str: Space key or "No space found" if space doesn't exist.
+        """
+        result = self.confluence.get_all_spaces()
+        spaces = result["results"]
+
+        for space in spaces:
+            if space["name"] == space_name:
+                logger.info(f"Found space key for '{space_name}'")
+                return space["key"]
+
+        logger.warning(f"No space named {space_name} found")
+        return "No space found"
+
+    def get_all_page_from_space(self, space_name: str):
+        """Retrieve all pages from a specific Confluence space.
+
+        Args:
+            space_name (str): Name of the Confluence space.
+
+        Returns:
+            list: Details of pages in the specified space.
+        """
+        logger.info(f"Retrieving all pages from space '{space_name}'")
+        space_key = self.get_space_key(space_name)
+        page_details = self.confluence.get_all_pages_from_space(
+            space_key, status=None, expand=None, content_type="page"
+        )
+        page_details = str([{"id": page["id"], "title": page["title"]} for page in page_details])
+        return page_details
+
+    def create_page(self, space_name: str, title: str, body: str, parent_id: Optional[str] = None) -> str:
+        """Create a new page in Confluence.
+
+        Args:
+            space_name (str): Name of the Confluence space.
+            title (str): Title of the new page.
+            body (str): Content of the new page.
+            parent_id (str, optional): ID of the parent page if creating a child page. Defaults to None.
+
+        Returns:
+            str: JSON-encoded page ID and title or error message.
+        """
+        try:
+            space_key = self.get_space_key(space_name=space_name)
+            page = self.confluence.create_page(space_key, title, body, parent_id=parent_id)
+            logger.info(f"Page created: {title} with ID {page['id']}")
+            return json.dumps({"id": page["id"], "title": title})
+        except Exception as e:
+            logger.error(f"Error creating page '{title}': {e}")
+            return json.dumps({"error": str(e)})
+
+    def update_page(self, page_id: str, title: str, body: str) -> str:
+        """Update an existing Confluence page.
+
+        Args:
+            page_id (str): ID of the page to update.
+            title (str): New title for the page.
+            body (str): Updated content for the page.
+
+        Returns:
+            str: JSON-encoded status and ID of the updated page or error message.
+        """
+        try:
+            updated_page = self.confluence.update_page(page_id, title, body)
+            logger.info(f"Page updated: {title} with ID {updated_page['id']}")
+            return json.dumps({"status": "success", "id": updated_page["id"]})
+        except Exception as e:
+            logger.error(f"Error updating page '{title}': {e}")
+            return json.dumps({"error": str(e)})

pyproject.toml

@@ -98,6 +98,7 @@ module = [
   "anthropic.*",
   "apify_client.*",
   "arxiv.*",
+  "atlassian.*",
   "boto3.*",
   "botocore.*",
   "bs4.*",