Confluence API Tool

[AhmadHakim2004]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

We would like to add a Confluence tool to integrate Confluence’s content management and collaboration capabilities into LangChain. This tool will offer a set of features for interacting with Confluence pages, attachments, templates, and comments by integrating the following API calls:

1. Search Using CQL
2. Page and Content Management
   - Create/Update Pages and Content
   - Get Page Content
   - Move Pages
   - Delete Pages
3. Template Management
   - Create/Update Templates
   - Get Templates
   - Delete Pages
4. Attach/Delete Attachments
5. Add/Get Comments
6. Get/Remove Content History

Motivation

After experimenting with Langchain, I noticed that there's a Confluence document loader, and a Jira toolkit that can be used for creating a confluence page indicating that Confluence is actively utilized by the Langchain community. And as someone who've used Confluence at work, I believe a separate and dedicated tool for Confluence with the functionalities mentioned above in LangChain would greatly enhance the experience with this platform. This would streamline workflows, improve content accessibility, and reduce manual interventions, making it easier to manage complex projects and maintain accurate documentation across our team.

Proposal (If applicable)

Here is a basic pesudocode implementation:

First we'll create an API wrapper:

class ConfluenceAPIWrapper(BaseModel):
    """Wrapper for Confluence API."""

    confluence: Any = None
    confluence_username: Optional[str] = None
    confluence_api_token: Optional[str] = None
    confluence_instance_url: Optional[str] = None

    @model_validator(mode="before")
    @classmethod
    def validate_environment(cls, values: Dict) -> Any:
        """Validate that api key and python package exists in environment."""
        confluence_username= get_from_dict_or_env(
            values, "confluence_username", "CONFLUENCE_USERNAME", default=""
        )
        values["confluence_username"] = confluence_username

        confluence_api_token= get_from_dict_or_env(
            values, "confluence_api_token", "CONFLUENCE_API_TOKEN"
        )
        values["confluence_api_token"] = confluence_api_token

        confluence_instance_url= get_from_dict_or_env(
            values, "confluence_instance_url", "CONFLUENCE_INSTANCE_URL"
        )
        values["confluence_instance_url"] = confluence_instance_url

        try:
            from atlassian import Confluence
        except ImportError:
            raise ImportError(
                "atlassian-python-api is not installed. "
                "Please install it with `pip install atlassian-python-api`"
            )

        confluence = Confluence(
            url=confluence_instance_url,
            username=confluence_username,
            password=confluence_api_token,
        )

        values["confluence"] = confluence
        return values

    def search(self, query: str) -> str:
        content = self.confluence.cql(query)
        parsed_content = self.parse_content(issues)
        return parsed_content

    def page_create(self, query: str) -> str:
        try:
            import json
        except ImportError:
            raise ImportError(
                "json is not installed. Please install it with `pip install json`"
            )
        params = json.loads(query)
        return self.confluence.create_page(**dict(params))

    . . .

    def run(self, mode: str, query: str) -> str:
        if mode == "cql":
            return self.search(query)
        elif mode == "create_page":
            return self.page_create(query)
        elif mode == "update_page":
            return self.page_update(query)
        elif mode == "delete_page":
            return self.page_delete(query)
        elif mode == "get_page_content":
            return self.get_page_content(query)
        elif mode == "update_page":
            return self.page_update(query)
        elif mode == "attach_content":
            return self.content_attach(query)
        elif mode == "attach_file":
            return self.file_attach(query)
        elif mode == "delete_attachment":
            return self.attachment_delete(query)
        else:
            raise ValueError(f"Got unexpected mode {mode}")

Then create tool class using the API wrapper:

class ConfluenceAction(BaseTool): 
    """Tool that queries the Atlassian Confluence API."""

    api_wrapper: ConfluenceAPIWrapper = Field(default_factory=ConfluenceAPIWrapper)
    mode: str
    name: str = ""
    description: str = ""

    def _run(
        self,
        instructions: str,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """Use the Atlassian ConfluenceAPI to run an operation."""
        return self.api_wrapper.run(self.mode, instructions)

Finally bundle the tools into a toolkit:

class ConfluenceToolkit(BaseToolkit):
    tools: List[BaseTool] = []

    @classmethod
    def from_confluence_api_wrapper(cls, confluence_api_wrapper: ConfluenceAPIWrapper) -> "ConfluenceToolkit":
        operations: List[Dict] = [
            {
                "mode": "cql",
                "name": "CQL Query",
                "description": CONFLUENCE_CQL_PROMPT,
            },
            {
                "mode": "create_page",
                "name": "Create confluence page",
                "description": CONFLUENCE_PAGE_CREATE_PROMPT,
            },
            {
                "mode": "update_page",
                "name": "Update confluence page",
                "description": CONFLUENCE_PAGE_UPDATE_PROMPT,
            },
            . . .
        ]

        tools = [
            ConfluenceAction(
                name=action["name"],
                description=action["description"],
                mode=action["mode"],
                api_wrapper=confluence_api_wrapper,
            )
            for action in operations
        ]
        return cls(tools=tools)

    def get_tools(self) -> List[BaseTool]:
        """Get the tools in the toolkit."""
        return self.tools

Example usage

os.environ["CONFLUENCE_API_TOKEN"] = "abc"
os.environ["CONFLUENCE_USERNAME"] = "123"
os.environ["CONFLUENCE_INSTANCE_URL"] = "https://confluence.atlassian.net"

from langchain_community.agent_toolkits import ConfluenceToolkit
toolkit = ConfluenceToolkit()

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
llm = ChatOpenAI(model="gpt-4o-mini")
agent_executor = create_react_agent(llm, toolkit.get_tools())

result = agent_executor.invoke({"input": "Find all pages in the Project Space that were created by Jane Doe after October 1, 2024.'"})

[AhmadHakim2004]

Hey @jacoblee93
We can raise a PR for this feature within a week. Do you think this is something we should work on and create an issue for?
Let us know if there are functionalities you want added/removed, we're open to implementing different ones.