GreyMatter AI ��� Multi-Agent LangGraph +

MCP Architecture Design

GreyMatter AI – Multi-Agent LangGraph + MCP Architecture Design

1. Overview
Problem
Solution
2. Architecture
Transport & Deployment
3. MCP Server Designs
3.1 JIRA MCP Server (NEW)
3.2 Confluence MCP Server (NEW)
3.3 Code Graph MCP Server – 5 New Tools
4. LangGraph Orchestrator
4.1 Graph Topology
4.2 State Schema
4.3 Entry Node
4.4 Router Node
4.5 Agent Nodes (ReAct Pattern)
4.6 Synthesizer Node
4.7 Exit Node
5. SSE Streaming Protocol
Current Protocol (PRESERVED – backward compatible)
New Event: agent_status (ADDITIVE)
Streaming Implementation
6. Error Handling & Resilience
Per-Agent Graceful Degradation
Health Checks
7. Frontend Changes (Minimal)
7.1 New Source Types in chat-ui/src/[Link]

7.2 SSE Handler Extension in chat-ui/src/[Link]

7.3 Code Graph Source Pills in chat-ui/src/[Link]
8. Configuration
orchestrator/[Link]
Environment Variables
9. File Structure
10. Docker Compose
11. Implementation Sequence
Sprint 1: MCP Servers (Week 1-2)
Sprint 2: LangGraph Orchestrator Core (Week 2-3)
Sprint 3: Integration + Frontend (Week 3-4)
Sprint 4: Polish (Week 4)
12. Verification Plan
Per-MCP-Server Testing
End-to-End Query Testing
Latency Targets
13. Key Decisions & Trade-offs
14. Risks & Mitigations
GreyMatter AI – Multi-Agent LangGraph + MCP
Architecture Design

1. Overview

Problem

GreyMatter AI has 4 knowledge sources spread across 3 independent modules:

Source Storage Module Current Access

Erlang code
Neo4j graph (9 layers) erlang_code_graph MCP server (10 tools)
structure

Function ChromaDB
erlang_code_graph MCP semantic_search tool
descriptions (erlang_function_summaries)

ChromaDB
JIRA tickets (jira_embeddings_12_10_t1) + rag-research Hybrid search inside RAGChat
REST API

ChromaDB
Confluence docs rag-research Hybrid search inside RAGChat
(confluence_embeddings_12_10_t1)

Users can only query JIRA + Confluence through the chat-ui. Code graph requires a separate MCP client. There is no
unified interface that can reason across all sources simultaneously.

Solution

An MCP-first multi-agent architecture where: - Every data source is an MCP server exposing well-defined tools
- A LangGraph orchestrator acts as an MCP client to all servers - An intent router classifies queries and activates
the right agent(s) - A synthesizer merges multi-source results into a coherent streamed response - The existing
chat-ui connects to the orchestrator with zero protocol changes

2. Architecture

+-----------------------+
| chat-ui |
| (React 19 + SSE) |
+----------+------------+
|
POST /chat/query/stream
|
+----------v------------+
| LangGraph |
| Orchestrator |
| (FastAPI + MCP |
| Client) |
+----------+------------+
|
+----------------+----------------+
| | |
+------v------+ +-----v-------+ +-----v--------+
| JIRA MCP | | Confluence | | Code Graph |
| Server | | MCP Server | | MCP Server |
| (NEW) | | (NEW) | | (EXISTING+5) |
+------+------+ +-----+-------+ +-----+--------+
| | |
+------v------+ +-----v-------+ +-----v--------+
| ChromaDB | | ChromaDB | | Neo4j + |
| (JIRA coll) | | (Confluence | | ChromaDB |
| + REST API | | coll) | | (summaries) |
+-------------+ +-------------+ +--------------+
Transport & Deployment

Component Transport Port Process

JIRA MCP Server SSE 8081 Standalone Python

Confluence MCP Server SSE 8082 Standalone Python

Code Graph MCP Server SSE 8080 Existing mcp_server.py

LangGraph Orchestrator HTTP (FastAPI) 8060 New FastAPI app

chat-ui HTTP 5713 React dev / Nginx

SSE transport chosen over stdio because: - Works over HTTP (Docker service discovery, health checks) - MCP SDK
supports it natively via sse_client() - Allows multiple clients (monitoring, testing) - Proper process isolation and
crash recovery

3. MCP Server Designs

3.1 JIRA MCP Server (NEW)

Location: mcp-servers/jira/jira_mcp_server.py

Tools

Tool Input Output Wraps

query: str, {success,

max_results: tickets[],
query_jira_with_llm() from rag-
search_jira int=5, jql_query,
research/src/providers/retrieval/jira_mcp_retrieval.py
include_details: reasoning,
bool=True total_found}

{metadata,
comments[],
get_ticket_details ticket_id: str extract_detailed_ticket_data() from same file
changelog[],
success}

{query, total,
query: str, results[{id, ChromaDB jira_embeddings_12_10_t1 collection vector
search_jira_semantic
top_k: int=10 document, metadata, search
relevance_score}]}

jql: str,
{success,
max_results:
execute_jql tickets[], Direct JIRA REST API via jira library
int=10, fields:
total_found}
list[str]

Implementation Notes

Uses FastMCP from [Link] (same pattern as existing code graph server)
Requires: JIRA_SERVER, JIRA_API_TOKEN, GEMINI_API_KEY, CHROMA_DB_HOST, CHROMA_DB_PORT

search_jira is the primary tool – it generates JQL via Gemini Flash, executes search, enriches results
search_jira_semantic provides vector-based fallback when JQL generation fails or for similarity queries
execute_jql is for advanced users / when the router agent wants precise control

Metadata returned per ticket (for UI source rendering)

 {
"source_type": "jira",
"ticket_id": "BUTL-12345",
"issue_key": "BUTL-12345",
"summary": "Handle race condition in order processing",
"status": "In Progress",
"priority": "High"
}

3.2 Confluence MCP Server (NEW)

Location: mcp-servers/confluence/confluence_mcp_server.py

Tools

Tool Input Output Wraps

{query, total,
query: str, results[{id,
top_k: document,
[Link]() from rag-
search_confluence int=10, metadata,
research/src/providers/retrieval/enhanced_hybrid_search.py
expand_query: rrf_score,
bool=False bm25_score,
semantic_score}]}

{query, total,
results[{id,
query: str, document, [Link]() from rag-
search_confluence_reranked
top_k: int=10 metadata, research/src/providers/retrieval/[Link]
rerank_score,
hybrid_score}]}

{id, document,
get_page_content page_id: str ChromaDB [Link](ids=[page_id])
metadata}

Implementation Notes

Loads ChromaDB collection and builds BM25 index at startup (30-60s one-time cost)
Uses text-embedding-004 (768-dim) for retrieval queries (matches existing collection)
search_confluence is the primary tool (BM25 + semantic + RRF + MMR)
search_confluence_reranked adds cross-encoder reranking (ms-marco-MiniLM-L-6-v2) for precision
Requires: CHROMA_DB_HOST, CHROMA_DB_PORT, GEMINI_API_KEY (for embeddings)

Metadata returned per page (for UI source rendering)

{
"source_type": "confluence",
"page_id": "123456789",
"space_key": "DEV",
"rrf_score": 0.85
}

3.3 Code Graph MCP Server – 5 New Tools

Location: erlang_code_graph/mcp_server.py (extend existing)

Existing 10 Tools (unchanged)

1. get_function_info(function) – Function details + callers/callees/storage

2. get_impact_analysis(function, depth) – What breaks if function changes
3. semantic_search(query, top_k) – NLP search on function descriptions
 4. get_call_chain(from_function, to_function) – Shortest path between functions
5. get_module_overview(module) – Module summary with dependencies
6. get_dependency_subgraph(target, direction, depth) – Graph neighborhood
7. get_functional_flow(function, depth) – Complete execution tree
8. analyze_change_impact(function, change_type, depth) – Risk scoring + recommendations
9. decompose_functionality(function) – Break into functional units
10. get_state_machine_flow(module) – FSM state diagram

New Tool 11: get_endpoints_for_query

@[Link]()
def get_endpoints_for_query(
query: str, # Module name, URL path fragment, or keyword
include_storage: bool = True # Include storage (Mnesia/ETS/Postgres) touched by endpoint
) -> str:
"""Find HTTP/REST endpoints defined in the codebase, optionally showing which databases they touch."""

Cypher query:

MATCH (e:Endpoint)<-[:EXPOSES]-(s:APIService)
WHERE [Link] CONTAINS $query OR e.handler_module CONTAINS $query
OPTIONAL MATCH (e)-[:HANDLED_BY]->(f:Function)-[:CALLS*1..3]->(f2:Function)-[r:READS_FROM|WRITES_TO]->(db)
RETURN [Link], [Link], e.handler_module, e.handler_function,
collect(DISTINCT {name: [Link], type: labels(db)[0], access: type(r)}) AS storage

Returns: {endpoints: [{path, method, handler_module, handler_function, storage_accessed[]}]}

New Tool 12: get_functions_by_storage

@[Link]()
def get_functions_by_storage(
storage_name: str, # Table/index name (e.g., "ppsbinrec", "order_items")
operation: str = "both" # "read", "write", or "both"
) -> str:
"""Find all functions that read from or write to a specific database table, ETS table, or Elastic index."""

Cypher query:

MATCH (f:Function)-[r:READS_FROM|WRITES_TO]->(db)
WHERE [Link] CONTAINS $storage_name
AND ($operation = 'both' OR
($operation = 'read' AND type(r) = 'READS_FROM') OR
($operation = 'write' AND type(r) = 'WRITES_TO'))
RETURN [Link], [Link], type(r) AS access_type, [Link] AS storage

Returns: {storage_name, operation, functions[{name, module, what, access_type}]}

New Tool 13: get_git_history_for_function

@[Link]()
def get_git_history_for_function(
function: str, # MFA string
max_commits: int = 20
) -> str:
"""Get git commit history for a specific function, with JIRA ticket extraction from commit messages."""

Neo4j query (Layer I):

MATCH (f:Function {name: $name, module: $module})<-[:HAS_VERSION]-(fv:FunctionVersion)<-[:INTRODUCED_BY]-(c:Commit)

RETURN [Link], [Link], [Link], [Link]
ORDER BY [Link] DESC LIMIT $max_commits

JIRA extraction: Regex [A-Z]+-\d+ on commit messages to extract ticket references.

Returns: {function, commits[{sha, message, author, date, jira_tickets[]}]}

New Tool 14: get_message_flow

@[Link]()
def get_message_flow(
target: str, # Module name or MFA string
include_kafka: bool = True
) -> str:
"""Trace internal Erlang message passing and Kafka topic flows for a module or function."""

Cypher queries (Layers E + G):

// Internal messaging
MATCH (f:Function {module: $module})-[:EMITS]->(mt:MessageType)
MATCH (f2:Function)-[:HANDLES_MSG]->(mt)
RETURN [Link], [Link] AS emitter, [Link] AS handler

// Kafka topics
MATCH (f:Function {module: $module})-[:PUBLISHES_TO]->(t:Topic)<-[:HAS_TOPIC]-(mb:MessageBroker)
RETURN [Link], [Link] AS publisher, [Link] AS broker

Returns: {target, messages_emitted[], messages_handled[], kafka_topics[{topic, role, module}]}

New Tool 15: hybrid_semantic_structural_search

@[Link]()
def hybrid_semantic_structural_search(
query: str, # Natural language description
module_filter: str = None, # Restrict to module
has_storage: str = None, # Must access this storage
is_otp_callback: bool = None, # Must be OTP callback
top_k: int = 10
) -> str:
"""Combine semantic search (ChromaDB) with Neo4j structural filters. Find functions matching a description that
also satisfy structural constraints."""

Implementation: 1. Run IntelligenceEngine.query_functions(query, n_results=top_k*3) for semantic candidates 2. For

each candidate, verify structural constraints via Neo4j: - Module filter: [Link] = $module_filter - Storage: (f)-

[:READS_FROM|WRITES_TO]->(:Database {name: $storage}) - OTP callback: (f)-[:IMPLEMENTS_CALLBACK]->(:OTPCallback) 3.

Return intersection, ranked by semantic score

Returns: {query, filters_applied, results[{function, module, what, relevance_score, structural_match}]}

4. LangGraph Orchestrator

4.1 Graph Topology

 +-----------+
| __start__|
+-----+-----+
|
+-----v------+
| entry | Load session, persist user message
+-----+------+
|
+-----v------+
| router | Classify intent, rewrite query, select servers
+-----+------+
|
+--------------+---------------+
| | | (parallel fan-out via Send())
+------v------+ +----v------+ +------v-------+
| jira_agent | |conf_agent | | code_agent |
| (ReAct, | |(ReAct, | | (ReAct, |
| MCP client)| | MCP client| | MCP client) |
+------+------+ +----+------+ +------+-------+
| | |
+--------------+---------------+
|
+-----v-------+
| synthesizer | Merge results, stream Gemini 2.5 Pro
+-----+-------+
|
+-----v------+
| exit | Persist assistant message, emit metadata
+-----+------+
|
+-----v-----+
| __end__ |
+-----------+

4.2 State Schema

# orchestrator/graph/[Link]

from typing import TypedDict, Optional

class AgentState(TypedDict):
# ── Input (set by entry node) ──
user_query: str
chat_id: str
user_email: str
chat_history: list[dict] # [{role, content}] from MySQL

# ── Router output ──
rewritten_query: str
intent: str # "jira" | "confluence" | "code" | "multi" | "general"
selected_servers: list[str] # ["jira", "confluence", "code"]
routing_reasoning: str

# ── Agent results (each agent writes its own key) ──

jira_results: Optional[dict] # {context: str, sources: list[dict]}
confluence_results: Optional[dict]
code_results: Optional[dict]

# ── Tool call traces (observability) ──

jira_tool_calls: list[dict] # [{tool_name, args, result_summary}]
confluence_tool_calls: list[dict]
code_tool_calls: list[dict]

# ── Synthesizer output ──
answer: str
metadata: list[dict] # Flat source list for UI
message_id: Optional[int]

# ── Error tracking ──
errors: list[dict] # [{server, error, timestamp}]

4.3 Entry Node

File: orchestrator/graph/nodes/[Link]

Responsibilities: 1. Create or validate chat session in MySQL (reuse MySQLConnection from rag-

research/src/core/database/mysql_client.py ) 2. Load conversation history from MySQL 3. Persist user message to DB 4.

Populate state.chat_id , state.chat_history

4.4 Router Node

File: orchestrator/graph/nodes/[Link]

Single LLM call to Gemini 2.5 Flash (fast, cheap) with structured JSON output:

Router Prompt

You are a query router for a knowledge platform with three data sources:

1. **JIRA** - Bug tickets, feature requests, issues, sprints, epics

Trigger keywords: ticket IDs (BUTL-XXX, GM-XXX), bug, issue, defect, sprint, epic, status, priority, assignee

2. **Confluence** - Documentation, architecture, processes, runbooks, how-to guides

Trigger keywords: documentation, wiki, architecture, design, process, how to, runbook, onboarding, guide

3. **Code Graph** - Erlang/OTP codebase structure, functions, modules, call chains, state machines, impact analysis
Trigger keywords: function, module, handle_call, gen_server, OTP, Erlang, call chain, impact, state machine,
database table, API endpoint, code

Given the user query and recent conversation, determine:

- Which data sources to query (can be multiple)
- A rewritten query optimized for search

User query: {query}

Recent conversation: {last_3_messages}

Return JSON:
{
"intent": "jira|confluence|code|multi|general",
"selected_servers": ["jira", "confluence", "code"],
"rewritten_query": "optimized search query",
"reasoning": "brief explanation"
}

Conditional Routing Logic

def route_to_agents(state: AgentState) -> list[Send]:

sends = []
for server in state["selected_servers"]:
if server == "jira":
[Link](Send("jira_agent", state))
elif server == "confluence":
[Link](Send("confluence_agent", state))
elif server == "code":
[Link](Send("code_agent", state))
if not sends:
[Link](Send("synthesizer", state)) # General/conversational
return sends

4.5 Agent Nodes (ReAct Pattern)

Each agent node: 1. Connects to its MCP server via SSE client session 2. Dynamically discovers available tools via
session.list_tools() 3. Uses a ReAct loop (LLM decides which tool to call, processes result, decides next action) 4.
Max 3 iterations to bound latency 5. Returns structured results + tool call trace

MCP Client Integration

 # orchestrator/mcp_client/session_manager.py

from mcp import ClientSession

from [Link] import sse_client

class MCPSessionManager:
"""Manages persistent MCP SSE client sessions to all servers."""

def __init__(self, server_configs: dict):

[Link] = server_configs # {name: {url, timeout}}
[Link]: dict[str, ClientSession] = {}

async def get_session(self, server_name: str) -> ClientSession:

if server_name not in [Link]:
url = [Link][server_name]["url"]
read, write = await sse_client(url)
session = ClientSession(read, write)
await [Link]()
[Link][server_name] = session
return [Link][server_name]

async def call_tool(self, server_name: str, tool_name: str, arguments: dict) -> str:
session = await self.get_session(server_name)
result = await session.call_tool(tool_name, arguments)
return [Link][0].text # MCP returns TextContent

LangChain Tool Adapter

# orchestrator/mcp_client/tool_adapter.py

from langchain_core.tools import BaseTool

class MCPToolAdapter(BaseTool):
"""Wraps an MCP tool as a LangChain tool for use in LangGraph agents."""

name: str
description: str
mcp_session_manager: MCPSessionManager
server_name: str
mcp_tool_name: str
input_schema: dict # JSON Schema from MCP tool definition

async def _arun(self, **kwargs) -> str:

return await self.mcp_session_manager.call_tool(
self.server_name, self.mcp_tool_name, kwargs
)

Agent Node Pattern (example: JIRA agent)

# orchestrator/graph/nodes/jira_agent.py

async def jira_agent_node(state: AgentState, config: dict) -> dict:

if "jira" not in state["selected_servers"]:
return {"jira_results": None, "jira_tool_calls": []}

session_manager = config["configurable"]["mcp_sessions"]
tools = await get_tools_for_server(session_manager, "jira")

# ReAct loop: LLM decides which tool(s) to call

agent = create_react_agent(
model=gemini_flash,
tools=tools,
max_iterations=3,
)

result = await [Link]({

"input": state["rewritten_query"],
"chat_history": state["chat_history"][-3:]
})

# Extract tool calls + format results

return {
"jira_results": {"context": result["output"], "sources": extract_sources(result)},
"jira_tool_calls": extract_tool_trace(result)
}

4.6 Synthesizer Node

File: orchestrator/graph/nodes/[Link]

The synthesizer is the only node that calls the main LLM (Gemini 2.5 Pro) and the only node that streams.

Multi-Source Prompt

Previous conversation:
{chat_history}

=== JIRA Tickets ===

{jira_context or "No JIRA data retrieved."}

=== Confluence Knowledge Base ===

{confluence_context or "No Confluence data retrieved."}

=== Code Intelligence (Erlang Codebase) ===

{code_context or "No code graph data retrieved."}

Abbreviations: {abbreviations}

User's question: {user_query}

Instructions:
- Synthesize information from ALL provided source sections
- When citing code structure, reference function names in module:function/arity format
- When citing documentation, mention the Confluence page
- When citing tickets, mention the JIRA issue key
- If a data source was unavailable, mention it briefly
- Use GitHub-Flavored Markdown formatting
- Be concise and direct

Metadata Assembly

The synthesizer merges sources from all agents into a flat list with source_type discriminant:

all_sources = []
if [Link]("jira_results"):
all_sources.extend(state["jira_results"]["sources"])
if [Link]("confluence_results"):
all_sources.extend(state["confluence_results"]["sources"])
if [Link]("code_results"):
all_sources.extend(state["code_results"]["sources"])

Source type values: - "jira" / "jira_mcp" – JIRA tickets (existing) - "confluence" – Confluence pages (existing) -
"code_graph_function" – Individual function info (new) - "code_graph_module" – Module overview (new) -
"code_graph_chain" – Call chain result (new) - "code_graph_impact" – Impact analysis (new)

4.7 Exit Node

File: orchestrator/graph/nodes/[Link]

1. Join all streamed chunks into full response

2. Persist assistant message to MySQL (returns message_id)

3. Return final {metadata, message_id} for SSE emission

5. SSE Streaming Protocol

Current Protocol (PRESERVED – backward compatible)

data: {"chat_id": "uuid-session-id"}

data: {"chunk": "Based on the "}
data: {"chunk": "code analysis..."}
data: {"metadata": [...], "message_id": 42}
data: [DONE]

New Event: agent_status (ADDITIVE)

 data: {"chat_id": "uuid-session-id"}
data: {"agent_status": {"agent": "jira", "status": "started"}}
data: {"agent_status": {"agent": "code_graph", "status": "started"}}
data: {"agent_status": {"agent": "jira", "status": "completed", "tools_called": ["search_jira"]}}
data: {"agent_status": {"agent": "code_graph", "status": "completed", "tools_called": ["semantic_search",
"get_function_info"]}}
data: {"agent_status": {"agent": "synthesizer", "status": "started"}}
data: {"chunk": "Based on the code analysis..."}
data: {"chunk": " the function handle_call/3..."}
data: {"metadata": [...], "message_id": 42}
data: [DONE]

The current chat-ui SSE parser ( [Link] lines 324-351) only checks for [Link] , [Link] , parsed.chat_id ,
[Link] , parsed.message_id . Unknown keys like agent_status are silently ignored. Zero frontend changes
required for basic functionality.

Streaming Implementation

# orchestrator/[Link]

@[Link]("/chat/query/stream")
async def query_stream(request: ChatQueryRequest, user = Depends(get_current_user)):
async def generate():
# 1. Entry: session setup
session_id = await ensure_session(request.chat_id, [Link])
yield sse_event({"chat_id": session_id})

# 2. Build and run graph

initial_state = {
"user_query": [Link],
"chat_id": session_id,
"user_email": [Link],
# ... other fields
}

# 3. Stream events from LangGraph

async for event in graph.astream_events(initial_state, version="v2"):
if event["event"] == "on_custom_event":
data = event["data"]
if [Link]("type") == "agent_status":
yield sse_event({"agent_status": data["status"]})
elif [Link]("type") == "chunk":
yield sse_event({"chunk": data["text"]})
elif [Link]("type") == "metadata":
yield sse_event({"metadata": data["sources"], "message_id": data["message_id"]})

yield "data: [DONE]\n\n"

return StreamingResponse(generate(), media_type="text/event-stream")

6. Error Handling & Resilience

Per-Agent Graceful Degradation

When an MCP server is down or a tool call times out:

async def jira_agent_node(state, config):

try:
# ... MCP tool calls ...
return {"jira_results": results}
except (ConnectionRefusedError, [Link], Exception) as e:
[Link](f"JIRA MCP server unavailable: {e}")
return {
"jira_results": None,
"errors": [Link]("errors", []) + [{
"server": "jira",
"error": str(e),
"timestamp": [Link]().isoformat()
}]
}
The synthesizer handles missing sources gracefully: - If 1 of 3 agents fails: synthesize from available sources, note
the gap - If all agents fail: return conversational response with error note - Never crash the entire pipeline for a
single agent failure

Health Checks

Each MCP server exposes a health endpoint. The orchestrator checks on startup and periodically:

# Startup health check

for server_name, config in mcp_servers.items():
try:
session = await session_manager.get_session(server_name)
tools = await session.list_tools()
[Link](f"{server_name}: {len(tools)} tools available")
except Exception as e:
[Link](f"{server_name}: unavailable ({e})")

7. Frontend Changes (Minimal)

7.1 New Source Types in chat-ui/src/[Link]

// Existing source types work unchanged: 'jira_mcp', 'jira', 'confluence'

// New code graph source types

export type CodeGraphSource = {
source_type: 'code_graph_function' | 'code_graph_module' | 'code_graph_chain' | 'code_graph_impact';
function_name?: string;
module?: string;
what?: string; // What-summary from ChromaDB
chain_description?: string;
affected_count?: number;
[key: string]: any;
};

// Agent status event type

export type AgentStatus = {
agent: string;
status: 'started' | 'completed' | 'error';
tools_called?: string[];
};

// Extended request
export type ChatRequest = {
query: string;
chat_id?: string;
include_jira_source?: boolean;
include_code_graph?: boolean; // NEW toggle
};

7.2 SSE Handler Extension in chat-ui/src/[Link]

Add 1 new handler and 2 lines in the parse loop:

// In postChatStream signature:
onAgentStatus?: (status: AgentStatus) => void;

// In SSE parse loop (after line 346):

if (parsed.agent_status) {
[Link]?.(parsed.agent_status);
}

7.3 Code Graph Source Pills in chat-ui/src/[Link]

Add includeCodeGraph state + checkbox toggle

Add code graph source deduplication (filter by source_type.startsWith('code_graph'))
Render code graph source pills with indigo/purple color and Code icon
Enhance LoadingMessage to show real agent names from agent_status events
8. Configuration

orchestrator/[Link]

mcp_servers:
jira:
url: "[Link]
enabled: true
timeout_seconds: 30
confluence:
url: "[Link]
enabled: true
timeout_seconds: 30
code_graph:
url: "[Link]
enabled: true
timeout_seconds: 30

langgraph:
router_model: "gemini-2.5-flash"
synthesis_model: "gemini-2.5-pro"
max_agent_iterations: 3
parallel_agents: true
query_rewriting: true

session:
mysql_host: "${MYSQL_DB_HOST}"
mysql_port: 3306
mysql_user: "${MYSQL_DB_USER}"
mysql_password: "${MYSQL_DB_PASSWORD}"
mysql_database: "butler_ai"

auth:
jwt_secret: "${JWT_SECRET}"
allowed_email_domain: "[Link]"

Environment Variables

# MCP Server URLs (override yaml)

MCP_JIRA_URL=[Link]
MCP_CONFLUENCE_URL=[Link]
MCP_CODE_GRAPH_URL=[Link]

# LLM
GEMINI_API_KEY=...

# JIRA MCP Server

JIRA_SERVER=[Link]
JIRA_API_TOKEN=...

# Code Graph MCP Server

NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=...
CHROMA_DB_HOST=[Link]
CHROMA_DB_PORT=8000

# Session DB
MYSQL_DB_HOST=[Link]
MYSQL_DB_PASSWORD=...
JWT_SECRET=...

9. File Structure
 greymatter-ai/
mcp-servers/ # NEW
jira/
jira_mcp_server.py # FastMCP server (4 tools)
[Link]
Dockerfile
.[Link]
confluence/
confluence_mcp_server.py # FastMCP server (3 tools)
[Link]
Dockerfile
.[Link]

orchestrator/ # NEW
[Link] # FastAPI entry point
[Link] # Config loader
[Link] # Configuration
graph/
__init__.py
[Link] # AgentState TypedDict
[Link] # StateGraph assembly
[Link] # Conditional routing logic
nodes/
__init__.py
[Link] # Session + user message persistence
[Link] # Intent classifier (Gemini Flash)
jira_agent.py # JIRA MCP client ReAct agent
confluence_agent.py # Confluence MCP client ReAct agent
code_agent.py # Code Graph MCP client ReAct agent
[Link] # Multi-source prompt + Gemini Pro streaming
[Link] # Assistant message persistence
mcp_client/
__init__.py
session_manager.py # MCP SSE client lifecycle
tool_adapter.py # MCP tool -> LangChain BaseTool
prompts/
__init__.py
router_prompt.py # Intent classification prompt
synthesis_prompt.py # Multi-source response prompt
agent_prompts.py # Per-agent ReAct system prompts
[Link]
Dockerfile
.[Link]

erlang_code_graph/
mcp_server.py # MODIFIED: add 5 new tools (#11-#15)

rag-research/ # KEPT (MCP servers import from here)

src/providers/retrieval/ # Imported by JIRA + Confluence MCP servers
src/core/database/ # Imported by orchestrator for MySQL

chat-ui/
src/[Link] # MODIFIED: add CodeGraphSource, AgentStatus
src/[Link] # MODIFIED: add onAgentStatus handler (2 lines)
src/[Link] # MODIFIED: add code graph toggle + source pills

[Link] # NEW: orchestrate all services

10. Docker Compose

 version: "3.8"
services:
jira-mcp:
build: ./mcp-servers/jira
ports: ["8081:8081"]
env_file: ./mcp-servers/jira/.env
healthcheck:
test: ["CMD", "curl", "-f", "[Link]
interval: 30s

confluence-mcp:
build: ./mcp-servers/confluence
ports: ["8082:8082"]
env_file: ./mcp-servers/confluence/.env
volumes:
- ./rag-research/databases/chromadb:/data/chromadb:ro
healthcheck:
test: ["CMD", "curl", "-f", "[Link]
interval: 30s

code-graph-mcp:
build: ./erlang_code_graph
ports: ["8080:8080"]
command: ["python", "mcp_server.py", "--transport", "sse", "--port", "8080"]
env_file: ./erlang_code_graph/.env
healthcheck:
test: ["CMD", "curl", "-f", "[Link]
interval: 30s

orchestrator:
build: ./orchestrator
ports: ["8060:8060"]
env_file: ./orchestrator/.env
depends_on:
jira-mcp: { condition: service_healthy }
confluence-mcp: { condition: service_healthy }
code-graph-mcp: { condition: service_healthy }

chat-ui:
build: ./chat-ui
ports: ["5713:5713"]
depends_on: [orchestrator]

11. Implementation Sequence

Sprint 1: MCP Servers (Week 1-2)

Step Task Deliverable

1.1 Create JIRA MCP Server mcp-servers/jira/jira_mcp_server.py with 4 tools

Create Confluence MCP

1.2 mcp-servers/confluence/confluence_mcp_server.py with 3 tools
Server

Add 5 new tools to Code

1.3 Extended mcp_server.py (tools #11-#15)
Graph MCP Server

Test each MCP server

1.4 mcp dev or direct SSE testing with curl
independently

Sprint 2: LangGraph Orchestrator Core (Week 2-3)

Step Task Deliverable

MCP client session manager

2.1 orchestrator/mcp_client/
+ tool adapter

State schema + entry/exit

2.2 orchestrator/graph/[Link], nodes/[Link], nodes/[Link]
nodes

Router node with intent

2.3 orchestrator/graph/nodes/[Link]
classification
 2.4 Agent nodes (JIRA, orchestrator/graph/nodes/{jira,confluence,code}_agent.py
Confluence, Code)

2.5 Synthesizer with streaming orchestrator/graph/nodes/[Link]

Graph builder + FastAPI

2.6 orchestrator/graph/[Link], orchestrator/[Link]
endpoints

Sprint 3: Integration + Frontend (Week 3-4)

Step Task Deliverable

Point chat-ui at new

3.1 Config change only
orchestrator

Add code graph source types

3.2 [Link], [Link], [Link] changes
to frontend

3.3 Add agent status display LoadingMessage enhancement

Docker compose for all

3.4 [Link]
services

End-to-end testing across all

3.5 Manual + automated
query types

Sprint 4: Polish (Week 4)

Step Task Deliverable

Error handling and graceful

4.1 Per-agent error boundaries
degradation

Timeouts and circuit

4.2 30s per-agent timeout
breakers

4.3 Performance tuning Parallel agents, token budgets

4.4 Production deployment Docker images, env configs

12. Verification Plan

Per-MCP-Server Testing

# JIRA MCP Server

python mcp-servers/jira/jira_mcp_server.py --transport sse --port 8081
# Test: curl to call search_jira tool via MCP protocol

# Confluence MCP Server

python mcp-servers/confluence/confluence_mcp_server.py --transport sse --port 8082
# Test: curl to call search_confluence tool

# Code Graph MCP Server (with new tools)

cd erlang_code_graph && python mcp_server.py --transport sse --port 8080
# Test: call new tools get_endpoints_for_query, get_functions_by_storage

End-to-End Query Testing

Query Expected Route Expected Sources

“What does
handle_call/3 in code code_graph_function
butler_server do?”

“Show me recent
jira jira tickets
critical bugs in BSS”
“How does the
confluence confluence pages
deployment process
work?”

“What does the

fulfillment module
multi (code + jira) code_graph_module + jira tickets
do and are there
open bugs?”

“Which functions
touch the
order_items table
and is there multi (code + confluence) code_graph functions + confluence pages
documentation
about the order
flow?”

“Hi, can you help

general none (conversational)
me?”

Latency Targets

Metric Target Notes

Time to first SSE event (chat_id) < 200ms Entry node only

Time to first agent_status event < 1s Router classification

Time to first text chunk < 5s (single agent), < 8s (multi) Includes MCP tool calls + synthesis

Total response time < 15s For typical queries

13. Key Decisions & Trade-offs

Decision Rationale Trade-off

SSE transport for MCP (not stdio) HTTP-based, multi-client, Docker-friendly Slightly more setup than stdio

Separate processes per MCP server Fault isolation, independent scaling Process management overhead

ReAct pattern for agents (not single Agents can make multi-step queries
Adds 1-2 LLM calls per agent
tool call) (search -> detail)

Gemini Flash for router + agents Fast (< 0.5s), cheap Less capable than Pro for edge cases

Gemini Pro for synthesis only Full capability where it matters most Single expensive call

Backward compatible with existing UI

Flat metadata array (not nested) Less structured
rendering

New orchestrator (not modifying rag- Clean separation, no risk to existing

Code duplication for auth/session
research) system

14. Risks & Mitigations

Risk Impact Mitigation

BM25 index build time in

Slow server startup Lazy build on first query; cache to disk
Confluence MCP (30-60s)

MCP session lifecycle

Connection drops Auto-reconnect in session_manager; health checks
management

30s timeout per agent; agent_status events keep user

Parallel agent latency > 8s Poor UX
informed
Neo4j connection from new
New dependency for MCP server Code Graph MCP server already handles this
process

Migration period (two

Confusion Feature flag in chat-ui config to switch orchestrator URL
backends running)

Gemini API rate limits under

Throttled responses Queue + backoff; separate API keys per server
load