How to Integrate AI Agents with Confluence

# Store credentials as environment variables
export CONFLUENCE_URL="https://your-org.atlassian.net/wiki"
export CONFLUENCE_EMAIL="you@company.com"
export CONFLUENCE_API_TOKEN="your-atlassian-api-token"

export CONFLUENCE_URL="https://confluence.your-company.com"
export CONFLUENCE_PAT="your-personal-access-token"

import requests
import os

auth = (os.getenv("CONFLUENCE_EMAIL"), os.getenv("CONFLUENCE_API_TOKEN"))
response = requests.get(
    f"{os.getenv('CONFLUENCE_URL')}/rest/api/space",
    auth=auth
)
print(response.status_code)  # Should be 200

pip install langchain langchain-openai atlassian-python-api python-dotenv

import os
from atlassian import Confluence
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

confluence = Confluence(
    url=os.getenv("CONFLUENCE_URL"),
    username=os.getenv("CONFLUENCE_EMAIL"),
    password=os.getenv("CONFLUENCE_API_TOKEN"),
    cloud=True  # Set to False for Server/DC
)


@tool
def search_confluence(query: str, space_key: str = None) -> str:
    """
    Search Confluence for pages matching the query.
    Optionally restrict to a specific space by space_key (e.g., 'ENG', 'HR').
    Returns page titles, URLs, and excerpts.
    """
    cql = f'text ~ "{query}"'
    if space_key:
        cql += f' AND space.key = "{space_key}"'
    cql += ' ORDER BY lastmodified DESC'

    results = confluence.cql(cql, limit=5, expand="excerpt")
    pages = results.get("results", [])

    if not pages:
        return f"No Confluence pages found for query: '{query}'"

    formatted = []
    for page in pages:
        title = page.get("title", "Untitled")
        url = f"{os.getenv('CONFLUENCE_URL')}{page.get('_links', {}).get('webui', '')}"
        excerpt = page.get("excerpt", "No excerpt available")
        formatted.append(f"**{title}**\nURL: {url}\nExcerpt: {excerpt[:300]}")
    return "\n\n".join(formatted)


@tool
def get_confluence_page_content(page_id: str) -> str:
    """Get the full text content of a Confluence page by its page ID."""
    page = confluence.get_page_by_id(page_id, expand="body.storage")
    if not page:
        return f"Page {page_id} not found"

    title = page.get("title", "Untitled")
    # Extract plain text from storage format
    body = page.get("body", {}).get("storage", {}).get("value", "")

    # Simple HTML tag removal for readable text
    import re
    clean_text = re.sub(r'<[^>]+>', '', body)
    clean_text = re.sub(r'\s+', ' ', clean_text).strip()

    return f"Page: {title}\n\nContent:\n{clean_text[:4000]}"


@tool
def create_confluence_page(space_key: str, title: str, content: str,
                           parent_page_title: str = None) -> str:
    """
    Create a new Confluence page. content should be plain text or simple HTML.
    parent_page_title optionally nests the page under an existing page.
    """
    parent_id = None
    if parent_page_title:
        parent = confluence.get_page_by_title(space_key, parent_page_title)
        if parent:
            parent_id = parent["id"]

    # Wrap content in basic HTML for Confluence storage format
    html_content = f"<p>{content.replace(chr(10), '</p><p>')}</p>"

    result = confluence.create_page(
        space=space_key,
        title=title,
        body=html_content,
        parent_id=parent_id,
        type='page'
    )

    page_url = f"{os.getenv('CONFLUENCE_URL')}{result.get('_links', {}).get('webui', '')}"
    return f"Page created: '{title}' — {page_url}"


@tool
def update_confluence_page(page_id: str, new_content: str) -> str:
    """Update the content of an existing Confluence page."""
    page = confluence.get_page_by_id(page_id)
    if not page:
        return f"Page {page_id} not found"

    current_version = page["version"]["number"]
    html_content = f"<p>{new_content.replace(chr(10), '</p><p>')}</p>"

    confluence.update_page(
        page_id=page_id,
        title=page["title"],
        body=html_content,
        version_number=current_version + 1
    )
    return f"Page {page_id} updated successfully"


@tool
def get_space_pages(space_key: str, limit: int = 20) -> str:
    """List pages in a Confluence space."""
    pages = confluence.get_all_pages_from_space(space_key, start=0, limit=limit)
    if not pages:
        return f"No pages found in space {space_key}"

    formatted = [f"Pages in {space_key} ({len(pages)} shown):"]
    for page in pages:
        title = page.get("title", "Untitled")
        page_id = page.get("id", "")
        formatted.append(f"- {title} (ID: {page_id})")
    return "\n".join(formatted)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
tools = [search_confluence, get_confluence_page_content,
         create_confluence_page, update_confluence_page, get_space_pages]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a knowledgeable assistant with access to the company's Confluence knowledge base.

When answering questions:
1. Search Confluence for relevant pages using specific keywords
2. Read the full content of the most relevant pages
3. Synthesize information into a clear, accurate answer
4. Always cite the Confluence pages you used
5. If information is outdated or unclear, say so

When creating documentation:
1. Use clear, professional language
2. Structure with appropriate headings and bullet points
3. Include relevant cross-references to existing Confluence pages"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

# Example: Employee onboarding question
result = executor.invoke({
    "input": "What is the process for requesting access to the production database?"
})
print(result["output"])

from langchain_community.document_loaders import ConfluenceLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# Load all pages from a Confluence space
loader = ConfluenceLoader(
    url=os.getenv("CONFLUENCE_URL"),
    username=os.getenv("CONFLUENCE_EMAIL"),
    api_key=os.getenv("CONFLUENCE_API_TOKEN"),
    space_key="ENG",
    include_attachments=False,
    limit=50
)

documents = loader.load()
print(f"Loaded {len(documents)} Confluence pages")

# Chunk and embed for vector search
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("confluence_index")

# Query the index
query = "How do I set up local development environment?"
relevant_docs = vectorstore.similarity_search(query, k=5)
for doc in relevant_docs:
    print(f"Source: {doc.metadata.get('source', 'Unknown')}")
    print(doc.page_content[:500])
    print("---")