flowchart TB
    subgraph Client["MCP Client"]
        C[Claude Desktop / MCP Client]
    end

    subgraph McpServer["McpServer (Facade)"]
        direction TB
        MS[McpServer]
    end

    subgraph Layers["Core Layers"]
        direction TB

        subgraph Session["Session Layer"]
            SM[SessionManager]
            SS[(SessionStore)]
        end

        subgraph Transport["Transport Layer"]
            TF[TransportFactory]
            SSE[SSETransport]
            HTTP[StreamableHttpTransport]
        end

        subgraph Execution["Execution Layer"]
            EC[ExecutionCoordinator]
            DS[DirectStrategy]
            QS[QueuedStrategy]
            PM[PendingCallsManager]
        end

        subgraph Protocol["Protocol Layer"]
            MP[MessageParser]
            MF[MessageFormatter]
        end
    end

    C <-->|SSE/HTTP| MS
    MS --> SM
    MS --> TF
    MS --> EC
    MS --> MP
    MS --> MF
    SM --> SS
    TF --> SSE
    TF --> HTTP
    EC --> DS
    EC --> QS
    QS --> PM

const mcpServer = McpServer.instance(logger);

flowchart TB
    subgraph Incoming["Incoming Requests"]
        GET["GET /sse (SSE setup)"]
        POST_INIT["POST /mcp (Streamable HTTP init)"]
        POST_MSG["POST /messages (tool call)"]
        DELETE["DELETE /mcp (session close)"]
    end

    subgraph McpServer["McpServer Facade"]
        HandleSetup[handleSetupRequest]
        HandleStreamable[handleStreamableHttpSetup]
        HandlePost[handlePostMessage]
        HandleDelete[handleDeleteRequest]
        HandleWorker[handleWorkerResponse]
        StorePending[storePendingResponse]
    end

    subgraph Internal["Internal Coordination"]
        CreateServer[createServer]
        SetupSession[setupSession]
        SetupHandlers[setupHandlers]
        CleanupSession[cleanupSession]
        RecreateTransport[recreateStreamableHttpTransport]
    end

    GET --> HandleSetup
    POST_INIT --> HandleStreamable
    POST_MSG --> HandlePost
    DELETE --> HandleDelete

    HandleSetup --> CreateServer
    HandleSetup --> SetupSession
    HandleStreamable --> CreateServer
    HandleStreamable --> SetupHandlers
    HandlePost --> RecreateTransport
    HandleDelete --> CleanupSession

    SetupSession --> SetupHandlers

interface HandlePostResult {
  wasToolCall: boolean;              // Whether the request was a tool call
  toolCallInfo?: McpToolCallInfo;    // Info about the tool call (if any)
  messageId?: string;                // The JSONRPC message ID
  relaySessionId?: string;           // Session ID for relayed requests (queue mode)
  needsListToolsRelay?: boolean;     // Whether this is a list-tools request needing relay
}

sequenceDiagram
    participant Client
    participant Main as McpServer (Main)
    participant Redis
    participant Worker

    Client->>Main: POST /messages (tool call)
    Main->>Main: storePendingResponse()
    Main-->>Client: 202 Accepted
    Main->>Redis: Enqueue job

    Redis->>Worker: Dequeue job
    Worker->>Worker: Execute tool
    Worker->>Redis: Publish result

    Redis->>Main: mcp-response event
    Main->>Main: handleWorkerResponse()
    Main-->>Client: Result via SSE/HTTP

flowchart LR
    subgraph Incoming["Incoming Request"]
        Raw["Raw JSON Body
{ jsonrpc: '2.0', method: 'tools/call', ... }"]
    end

    subgraph MessageParser["MessageParser"]
        Parse[parse]
        IsToolCall[isToolCall]
        IsListTools[isListToolsRequest]
        GetId[getRequestId]
        Extract[extractToolCallInfo]
    end

    subgraph Outgoing["Outgoing Response"]
        Result["Tool Execution Result
(string, object, Error)"]
    end

    subgraph MessageFormatter["MessageFormatter"]
        FormatResult[formatToolResult]
        FormatError[formatError]
    end

    Raw --> Parse
    Parse --> IsToolCall
    Parse --> IsListTools
    Parse --> GetId
    Parse --> Extract
    Extract --> Info["McpToolCallInfo
{ toolName, arguments }"]

    Result --> FormatResult
    Result --> FormatError
    FormatResult --> McpResult["McpToolResult
{ content: [{ type, text }] }"]
    FormatError --> McpResult

// Extracted info from a tool call request
interface McpToolCallInfo {
  toolName: string;                    // Name of the tool to invoke
  arguments: Record<string, unknown>;  // Arguments passed to the tool
  sourceNodeName?: string;             // Optional: n8n node that registered the tool
}

// Formatted result to send back to the client
interface McpToolResult {
  content: Array<{ type: string; text: string }>;  // MCP content blocks
  isError?: boolean;                                // Flag for error responses
}

// Special marker returned when handling list-tools requests
const MCP_LIST_TOOLS_REQUEST_MARKER = { _listToolsRequest: true };

// 1. Client sends a tool call
const body = '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_weather","arguments":{"city":"London"}}}';

// 2. Parse and identify
MessageParser.isToolCall(body);  // true
MessageParser.isListToolsRequest(body);  // false

// 3. Extract tool info
const info = MessageParser.extractToolCallInfo(body);
// { toolName: 'get_weather', arguments: { city: 'London' } }

// 4. Execute tool and format result
const result = await executeTool(info);  // { temperature: 15, unit: 'celsius' }
const formatted = MessageFormatter.formatToolResult(result);
// { content: [{ type: 'text', text: '{"temperature":15,"unit":"celsius"}' }] }

// 5. Or format an error
const error = MessageFormatter.formatError(new Error('City not found'));
// { isError: true, content: [{ type: 'text', text: 'Error: City not found' }] }

flowchart TB
    subgraph SessionManager["SessionManager (Coordinator)"]
        direction LR
        Register[registerSession]
        Destroy[destroySession]
        GetSession[getSession]
        GetTransport[getTransport]
        GetServer[getServer]
        IsValid[isSessionValid]
        Tools[getTools / setTools]
    end

    subgraph InMemoryState["In-Memory State (SessionManager)"]
        SessionInfo["sessions: Record<sessionId, SessionInfo>"]
        SI_Content["SessionInfo = { sessionId, server, transport }"]
    end

    subgraph SessionStore["SessionStore Interface"]
        InMemory[InMemorySessionStore]
        Redis[RedisSessionStore]
    end

    SessionManager --> InMemoryState
    SessionManager --> SessionStore
    InMemory -.->|implements| SessionStore
    Redis -.->|implements| SessionStore

interface SessionStore {
  register(sessionId: string): Promise<void>;      // Register a new session
  validate(sessionId: string): Promise<boolean>;   // Check if session exists
  unregister(sessionId: string): Promise<void>;    // Remove a session
  getTools(sessionId: string): Tool[] | undefined; // Get tools for session
  setTools(sessionId: string, tools: Tool[]): void; // Associate tools with session
  clearTools(sessionId: string): void;             // Remove tools from session
}

flowchart TB
    subgraph Client["MCP Client"]
        C[Claude Desktop / MCP Client]
    end

    subgraph TransportLayer["Transport Layer"]
        subgraph McpTransport["McpTransport Interface"]
            Send[send]
            HandleReq[handleRequest]
            Close[close]
        end

        subgraph Implementations["Implementations"]
            SSE[SSETransport]
            HTTP[StreamableHttpTransport]
        end

        subgraph Factory["TransportFactory"]
            CreateSSE[createSSE]
            CreateHTTP[createStreamableHttp]
            Recreate[recreateStreamableHttp]
        end
    end

    C <-->|"GET /sse + POST /messages"| SSE
    C <-->|"POST /mcp"| HTTP
    SSE -.->|implements| McpTransport
    HTTP -.->|implements| McpTransport
    Factory --> SSE
    Factory --> HTTP

interface McpTransport {
  readonly transportType: 'sse' | 'streamableHttp';  // Identifies the transport type
  readonly sessionId: string | undefined;             // Session ID for this connection

  send(message: JSONRPCMessage): Promise<void>;       // Send a message to the client
  handleRequest(req, resp, body?): Promise<void>;     // Handle an incoming request
  close?(): Promise<void>;                            // Close the transport

  onclose?: () => void | Promise<void>;               // Callback when connection closes
}

sequenceDiagram
    participant Client
    participant Server

    Client->>Server: GET /sse
    Note over Server: Connection stays open
    Server-->>Client: SSE: endpoint event (POST URL)

    Client->>Server: POST /messages (tool call)
    Server-->>Client: SSE: message event (result)

    Client->>Server: POST /messages (another call)
    Server-->>Client: SSE: message event (result)

    Note over Client,Server: Connection persists until closed

sequenceDiagram
    participant Client
    participant Server

    Client->>Server: POST /mcp (initialize)
    Server-->>Client: Response + mcp-session-id header

    Client->>Server: POST /mcp (tool call)
Header: mcp-session-id
    Server-->>Client: Response (result)

    Client->>Server: DELETE /mcp
Header: mcp-session-id
    Server-->>Client: 200 OK (session closed)

sequenceDiagram
    participant Client
    participant InstanceA as Instance A
    participant InstanceB as Instance B
    participant Redis

    Client->>InstanceA: POST /mcp (initialize)
    InstanceA->>Redis: Register session
    InstanceA-->>Client: mcp-session-id: abc123

    Note over Client,InstanceB: Load balancer routes to different instance

    Client->>InstanceB: POST /mcp (tool call)
Header: mcp-session-id: abc123
    InstanceB->>Redis: Validate session exists
    InstanceB->>InstanceB: recreateStreamableHttp(abc123)
    InstanceB-->>Client: Response

type CompressionResponse = Response & {
  flush?: () => void;
};

flowchart TB
    subgraph ExecutionCoordinator
        Execute[executeTool]
        SetStrategy[setStrategy]
    end

    subgraph Strategies["ExecutionStrategy Interface"]
        Direct[DirectExecutionStrategy]
        Queued[QueuedExecutionStrategy]
    end

    subgraph PendingCalls[PendingCallsManager]
        Wait[waitForResult]
        Resolve[resolve]
    end

    ExecutionCoordinator --> Strategies
    Direct -->|invoke| Tool[Tool.invoke]
    Queued --> PendingCalls
    PendingCalls -.->|worker response| Resolve

interface ExecutionStrategy {
  executeTool(
    tool: Tool,
    args: Record<string, unknown>,
    context: ExecutionContext,
  ): Promise<unknown>;
}

interface ExecutionContext {
  sessionId: string;
  messageId?: string;
}

const strategy = new DirectExecutionStrategy();
const result = await strategy.executeTool(tool, args, context);
// Directly calls tool.invoke(args)

const strategy = new QueuedExecutionStrategy(
  pendingCallsManager,
  timeoutMs  // Optional, defaults to 120000ms (2 minutes)
);

// Methods for resolving calls from workers:
strategy.resolveToolCall(callId, result);  // Returns true if call was pending
strategy.rejectToolCall(callId, error);    // Returns true if call was pending
strategy.getPendingCallsManager();         // Access the pending calls manager

const manager = new PendingCallsManager();

// Wait for a result (with timeout)
const result = await manager.waitForResult(callId, toolName, args, timeoutMs);

// Resolve/reject from worker
manager.resolve(callId, result);
manager.reject(callId, error);

// Query and manage pending calls
manager.has(callId);                    // Check if call is pending
manager.get(callId);                    // Get pending call info
manager.remove(callId);                 // Remove without resolving
manager.cleanupBySessionId(sessionId);  // Clean up all calls for a session

import { McpServer } from './McpServer';

const mcpServer = McpServer.instance(logger);

// Handle SSE setup
await mcpServer.handleSetupRequest(req, resp, serverName, postUrl, tools);

// Handle POST messages
const result = await mcpServer.handlePostMessage(req, resp, tools, serverName);

import { McpServer } from './McpServer';
import { QueuedExecutionStrategy } from './execution';
import { RedisSessionStore } from './RedisSessionStore';

const mcpServer = McpServer.instance(logger);

// Configure Redis session store
mcpServer.setSessionStore(new RedisSessionStore(publisher, getKey, ttl));

// Configure queued execution
mcpServer.setExecutionStrategy(
  new QueuedExecutionStrategy(mcpServer.getPendingCallsManager())
);

sequenceDiagram
    participant Client
    participant McpServer
    participant Transport
    participant Session

    Client->>McpServer: GET /sse (setup)
    McpServer->>Transport: createSSE()
    Transport-->>McpServer: SSETransport
    McpServer->>Session: registerSession()
    McpServer-->>Client: SSE stream opened

    Client->>McpServer: POST /messages (tool call)
    McpServer->>Session: getTransport()
    McpServer->>Transport: handleRequest()
    Note over McpServer: Execute tool
    Transport-->>Client: Tool result via SSE

sequenceDiagram
    participant Client
    participant Main
    participant McpServer
    participant Worker
    participant Redis

    Client->>Main: Tool call request
    Main->>McpServer: handlePostMessage()
    McpServer->>McpServer: storePendingResponse()
    Main->>Redis: Enqueue job
    Main-->>Client: 202 Accepted

    Redis->>Worker: Dequeue job
    Worker->>Worker: Execute tool
    Worker->>Redis: Publish result

    Redis->>Main: mcp-response event
    Main->>McpServer: handleWorkerResponse()
    McpServer-->>Client: Result via SSE

McpTrigger/
├── McpServer.ts              # Main facade coordinating all subsystems
├── McpTrigger.node.ts        # n8n node implementation
├── protocol/                 # JSONRPC message parsing & formatting
│   ├── MessageParser.ts
│   ├── MessageFormatter.ts
│   └── types.ts
├── session/                  # Client connection & state management
│   ├── SessionManager.ts
│   ├── SessionStore.ts
│   └── InMemorySessionStore.ts
├── transport/                # SSE & Streamable HTTP protocols
│   ├── Transport.ts
│   ├── SSETransport.ts
│   ├── StreamableHttpTransport.ts
│   └── TransportFactory.ts
├── execution/                # Direct & queued execution strategies
│   ├── ExecutionStrategy.ts
│   ├── DirectExecutionStrategy.ts
│   ├── QueuedExecutionStrategy.ts
│   ├── PendingCallsManager.ts
│   └── ExecutionCoordinator.ts
└── __tests__/                # Comprehensive unit tests

// Main facade
import { McpServer, MCP_LIST_TOOLS_REQUEST_MARKER } from './McpServer';
import type { HandlePostResult } from './McpServer';

// Protocol
import { MessageParser, MessageFormatter } from './protocol';
import type { McpToolCallInfo, McpToolResult } from './protocol';

// Session
import { InMemorySessionStore, SessionManager } from './session';
import type { SessionStore } from './session';

// Transport
import { SSETransport, StreamableHttpTransport, TransportFactory } from './transport';
import type { McpTransport, CompressionResponse, TransportType } from './transport';

// Execution
import { DirectExecutionStrategy, QueuedExecutionStrategy, PendingCallsManager, ExecutionCoordinator } from './execution';
import type { ExecutionStrategy, ExecutionContext } from './execution';