Build a Confluence Knowledge Base MCP Server for Agentic Document Discovery in 2026

The 4.7-Hour Weekly Search Tax

Enterprise teams spend 4.7 hours per week searching for documentation that already exists in Confluence. The problem: knowledge is scattered across 200+ spaces, 15,000 pages, and poorly named attachments. When an AI agent needs context to answer a question, it can't search Confluence without human intermediation.

This FastMCP server exposes Confluence as a set of MCP tools: pages can be retrieved by ID or title, CQL (Confluence Query Language) search is available for complex queries, spaces can be listed and explored, and new pages can be created from agent-generated content.

File 1: confluence_server.py — FastMCP Python Server

from fastmcp import FastMCP
import httpx
import os
from typing import Optional
import json

mcp = FastMCP(
    name="confluence-knowledge-base",
    version="1.0.0"
)

BASE_URL = os.environ.get("CONFLUENCE_URL", "https://yourcompany.atlassian.net")
AUTH = (os.environ.get("CONFLUENCE_EMAIL", ""), os.environ.get("CONFLUENCE_API_TOKEN", ""))

async def _get(path: str, params: dict = None) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/wiki/api/v2{path}",
            auth=AUTH,
            params=params or {},
            timeout=30)
        resp.raise_for_status()
        return resp.json()

async def _post(path: str, data: dict) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/wiki/api/v2{path}",
            auth=AUTH,
            json=data,
            timeout=30)
        resp.raise_for_status()
        return resp.json()

@mcp.tool()
async def get_page(page_id: str) -> str:
    """Retrieve a Confluence page by ID with its body content."""
    page = await _get(f"/pages/{page_id}", {"body-format": "storage"})
    result = {
        "id": page.get("id"),
        "title": page.get("title"),
        "space": page.get("space", {}).get("name", "Unknown"),
        "body": page.get("body", {}).get("storage", {}).get("value", "")[:5000],
        "url": f"{BASE_URL}/wiki/spaces/{page.get('space', {}).get('key', '')}/pages/{page.get('id')}"
    }
    return json.dumps(result, indent=2)

@mcp.tool()
async def search_pages(
    query: str,
    space_key: Optional[str] = None,
    limit: int = 10
) -> str:
    """Search Confluence using CQL (Confluence Query Language)."""
    cql = f"text ~ \"{query}\""
    if space_key:
        cql += f" AND space = {space_key}"
    cql += " ORDER BY lastmodified DESC"

    results = await _get("/content/search", {
        "cql": cql,
        "limit": min(limit, 25),
        "expand": "space,version"
    })

    pages = []
    for item in results.get("results", []):
        pages.append({
            "id": item.get("id"),
            "title": item.get("title"),
            "space": item.get("space", {}).get("name", "Unknown"),
            "url": item.get("_links", {}).get("base", "") + item.get("_links", {}).get("webui", ""),
            "excerpt": item.get("excerpt", "")[:200]
        })

    return json.dumps({"count": len(pages), "pages": pages}, indent=2)

@mcp.tool()
async def list_spaces(limit: int = 20) -> str:
    """List all Confluence spaces the agent has access to."""
    spaces = await _get("/spaces", {"limit": min(limit, 50)})
    space_list = []
    for s in spaces.get("results", []):
        space_list.append({
            "key": s.get("key"),
            "name": s.get("name"),
            "type": s.get("type"),
            "description": (s.get("description", {}).get("plain", {}).get("value", ""))[:100]
        })
    return json.dumps({"count": len(space_list), "spaces": space_list}, indent=2)

@mcp.tool()
async def create_page(
    space_key: str,
    title: str,
    body_storage: str,
    parent_id: Optional[str] = None
) -> str:
    """Create a new Confluence page with storage-format body."""
    payload = {
        "spaceId": space_key,
        "title": title,
        "body": {"storage": {"value": body_storage, "representation": "storage"}}
    }
    if parent_id:
        payload["parentId"] = parent_id

    result = await _post("/pages", payload)
    return json.dumps({
        "id": result.get("id"),
        "title": result.get("title"),
        "url": f"{BASE_URL}/wiki/spaces/{space_key}/pages/{result.get('id')}"
    }, indent=2)

if __name__ == "__main__":
    mcp.run(transport="stdio")

File 2: .env.example

CONFLUENCE_URL=https://yourcompany.atlassian.net
CONFLUENCE_EMAIL=you@company.com
CONFLUENCE_API_TOKEN=your-api-token-here

Claude Desktop Configuration

{
  "mcpServers": {
    "confluence": {
      "command": "python",
      "args": ["confluence_server.py"],
      "env": {
        "CONFLUENCE_URL": "https://yourcompany.atlassian.net",
        "CONFLUENCE_EMAIL": "you@company.com",
        "CONFLUENCE_API_TOKEN": "your-token"
      }
    }
  }
}

Production Reality Check

• Auth: Use Atlassian API tokens (not passwords); rotate every 90 days
• Rate limits: Confluence Cloud allows ~100 requests/minute; implement retry with backoff
• Body truncation: Large pages (>5000 chars) are truncated; use page_id for full content
• Permissions: Agent inherits the API token user's permissions; use a read-only token for search-only use cases
• Cost: Confluence API is included in all Cloud plans at no additional cost

Setup Commands

# Install dependencies
pip install fastmcp httpx

# Run the server
python confluence_server.py

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Discover more MCP integrations in our MCP Directory and explore Google Workspace MCP and Linear issue triage.

Last tested: August 2026 with Python 3.12, Node v22, and latest framework releases.