Using Purple8 with MCP Agents

Purple8 ships a first-party MCP server out of the box. One command exposes the full REST API as typed tools that any MCP-compatible client — Claude Desktop, Cursor, or your own agent loop — can discover and call via JSON-RPC 2.0 over stdio.

What MCP is (the 30-second version)

The Model Context Protocol is an open standard (originally from Anthropic, now widely adopted) that lets an AI agent discover and call tools through a uniform JSON-RPC 2.0 interface over stdio or HTTP-SSE.

From the agent's perspective, a tool is just a named function with a JSON Schema for its inputs. The agent decides when to call it; the MCP server executes the real work and returns a text result. The agent has no direct access to your database or API — it only sees what the tool returns.

Architecture

MCP Client (Claude Desktop / Cursor / custom agent)
  │  JSON-RPC 2.0 (stdio)
  ▼
purple8-graph mcp-server  ← ships with the package, pip install "purple8-graph[mcp]"
  │  REST + API-Key / JWT
  ▼
Purple8 Secure Server (http://localhost:8010)

---

Quick start — native MCP server

Step 1 — install the MCP extra

pip install "purple8-graph[mcp]"

Step 2 — start the server

With an API key (recommended for persistent servers):

purple8-graph mcp-server \
    --url     http://localhost:8010 \
    --api-key YOUR_API_KEY

With username / password (JWT, re-authenticated automatically on expiry):

purple8-graph mcp-server \
    --url      http://localhost:8010 \
    --username admin \
    --password changeme

Flag	Required	Default	Description
--url	✅	—	Base URL of the running Purple8 server
--api-key	✅ (or --username)	—	Long-lived API key (X-API-Key header)
--username	✅ (or --api-key)	—	Admin username for JWT login
--password	if --username set	—	Admin password for JWT login
--timeout	❌	30s	HTTP request timeout in seconds

Step 3 — connect Claude Desktop

Add this to ~/.claude/claude_desktop_config.json:

{
  "mcpServers": {
    "purple8": {
      "command": "purple8-graph",
      "args": [
        "mcp-server",
        "--url", "http://localhost:8010",
        "--api-key", "YOUR_API_KEY"
      ]
    }
  }
}

Restart Claude Desktop — the 13 Purple8 tools will appear in the tool chooser immediately.

Step 3 (alternative) — connect a custom Python agent

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

params = StdioServerParameters(
    command="purple8-graph",
    args=["mcp-server", "--url", "http://localhost:8010", "--api-key", "YOUR_API_KEY"],
)

async def run():
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Hybrid search
            result = await session.call_tool("hybrid_search", {
                "query_text": "enterprise contracts expiring Q3 2026",
                "label": "Contract",
                "k": 5,
            })
            print(result.content)

            # Ask a natural-language question
            answer = await session.call_tool("rag_query", {
                "question": "Which accounts have open critical tickets right now?",
            })
            print(answer.content)

asyncio.run(run())

---

Tools exposed by the native server

All 13 tools are available out of the box — no wrapper code needed.

Tool name	Method	Path	What it does
graph_query	POST	/query	Arbitrary Cypher
vector_search	POST	/search/vector	HNSW nearest-neighbour
hybrid_search	POST	/search/hybrid	Vector + graph combined
rag_query	POST	/rag/query	NL question → LLM answer
traverse	POST	/traverse	Graph walk from a node
add_node	POST	/nodes	Insert a node
get_node	GET	/nodes/{id}	Fetch a node
ingest_text	POST	/ingest/preview + /ingest/commit	Extract entities from text
journey_start	POST	/v1/journeys/{type}/instances	Start a Journey Engine instance
journey_advance	POST	/v1/journeys/instances/{id}/advance	Advance to next stage
journey_status	GET	/v1/journeys/instances/{id}	Full audit trail
pagerank	GET	/algorithms/pagerank	Top-K influential nodes
communities	GET	/algorithms/communities	Louvain community detection

---

Authentication details

Approach	How	When to use
API key	--api-key YOUR_KEY	Long-running servers, CI/CD
JWT	--username + --password (auto-refreshed on 401)	Per-session auth, scoped permissions

---

Building your own wrapper (advanced)

If you need custom tool names, filtered results, or want to embed the MCP logic inside a larger service, you can write a thin wrapper directly against the REST API instead of using the native server.

Install dependencies

pip install mcp httpx

Minimal wrapper

# purple8_mcp.py  —  custom wrapper, only if you need non-standard tool names/logic
import asyncio, json, httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

BASE_URL = "http://localhost:8010"
HEADERS  = {"X-API-Key": "YOUR_API_KEY"}
server   = Server("purple8-graph-custom")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="hybrid_search",
            description="Semantic + graph search over the Purple8 knowledge graph.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query_text": {"type": "string"},
                    "label":      {"type": "string"},
                    "k":          {"type": "integer", "default": 10},
                },
                "required": ["query_text"],
            },
        ),
        # add more tools as needed — see the tools table above
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    async with httpx.AsyncClient(headers=HEADERS, timeout=30) as client:
        if name == "hybrid_search":
            r = await client.post(f"{BASE_URL}/search/hybrid", json=arguments)
        else:
            return [TextContent(type="text", text=f"Unknown tool: {name}")]
        r.raise_for_status()
        return [TextContent(type="text", text=json.dumps(r.json(), indent=2))]

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

---

See Also

• Hybrid Search — vector + graph retrieval
• Journey Engine — journey types, stages, and audit trails
• Graph as Memory — how the graph accumulates AI decision history
• Encryption & KMS — what the agent can and cannot see