12 - Collaborazione Inter-Server e EventBus

Introduzione: Come Comunicano i Server MCP

In un ecosistema con 22 server MCP attivi, la collaborazione tra server diventa essenziale. Un singolo server gestisce un dominio specifico (scrum, time-tracking, CI/CD), ma i flussi di lavoro reali attraversano più domini: quando uno sprint si chiude, le metriche devono aggiornarsi, i costi devono calcolarsi, la retrospettiva deve crearsi automaticamente.

Tech-MCP implementa due meccanismi complementari per la comunicazione inter-server: l'EventBus (pub/sub asincrono fire-and-forget) e il ClientManager (chiamate RPC sincrone). Insieme, permettono di costruire flussi end-to-end che attraversano l'intera suite senza accoppiamento diretto tra i server.

EventBus: Il Cuore della Collaborazione Asincrona

L'EventBus e il meccanismo che permette ai server MCP Suite di collaborare in modo asincrono e disaccoppiato. Quando un server esegue un'azione significativa (creazione sprint, log tempo, completamento build), pubblica un evento tipizzato che altri server possono sottoscrivere per reagire automaticamente.

Server A                    EventBus                    Server B
   |                           |                           |
   |-- publish(evento) ------->|                           |
   |                           |-- handler(payload) ------>|
   |                           |                           |
   |   (fire-and-forget)       |   (delivery garantita     |
   |                           |    in-process)            |
    

Questo pattern e noto come Publish/Subscribe (Pub/Sub) e offre vantaggi fondamentali:

• Disaccoppiamento: il publisher non conosce i subscriber
• Estensibilita: aggiungere un nuovo subscriber non richiede modifiche al publisher
• Opzionalita: l'EventBus e sempre opzionale; ogni server funziona perfettamente anche senza

Interfaccia EventBus

L'interfaccia EventBus definisce il contratto per qualsiasi implementazione del bus eventi:

interface EventBus {
  publish<E extends EventName>(event: E, payload: EventPayload<E>): Promise<void>;
  subscribe<E extends EventName>(event: E, handler: EventHandler<E>): () => void;
  subscribePattern(pattern: string, handler: PatternHandler): () => void;
  clear(): void;
}
    

LocalEventBus: Implementazione In-Process

La classe LocalEventBus e l'implementazione predefinita, basata su Node.js EventEmitter con supporto pattern via micromatch:

import { EventEmitter } from "node:events";
import micromatch from "micromatch";

interface PatternSubscription {
  pattern: string;
  handler: PatternHandler;
}

export class LocalEventBus implements EventBus {
  private emitter = new EventEmitter();
  private patternSubs: PatternSubscription[] = [];

  constructor() {
    this.emitter.setMaxListeners(100);
  }

  async publish(event: string, payload: unknown): Promise<void> {
    // Notifica subscriber esatti
    this.emitter.emit(event, payload);

    // Notifica subscriber pattern
    for (const sub of this.patternSubs) {
      if (micromatch.isMatch(event, sub.pattern)) {
        try {
          await sub.handler(event, payload);
        } catch {
          // Errori nei subscriber non bloccano la pubblicazione
        }
      }
    }
  }

  subscribe(event: string, handler: EventHandler): () => void {
    this.emitter.on(event, handler);
    return () => this.emitter.off(event, handler);
  }

  subscribePattern(pattern: string, handler: PatternHandler): () => void {
    const sub: PatternSubscription = { pattern, handler };
    this.patternSubs.push(sub);
    return () => {
      const index = this.patternSubs.indexOf(sub);
      if (index >= 0) this.patternSubs.splice(index, 1);
    };
  }

  clear(): void {
    this.emitter.removeAllListeners();
    this.patternSubs = [];
  }
}
    

Caratteristiche chiave: max 100 listener per evento, pattern matching via micromatch per wildcard (scrum:*, *:completed), errori nei pattern handler catturati silenziosamente per non interrompere la pubblicazione, e zero dipendenze esterne oltre micromatch.

Tipizzazione Completa con EventMap

Ogni evento e tipizzato sia nel nome che nel payload grazie alla mappa EventMap. Il compilatore TypeScript verifica a compile-time che il payload sia corretto:

// Il compilatore TypeScript verifica che il payload sia corretto
eventBus.publish('scrum:sprint-started', {
  sprintId: '42',
  name: 'Sprint 15',
  startDate: '2025-01-01',
  endDate: '2025-01-14',
});

// Errore di compilazione: manca il campo 'name'
eventBus.publish('scrum:sprint-started', {
  sprintId: '42',
});
    

I 29 Eventi Tipizzati in 11 Domini

MCP Suite definisce 29 eventi tipizzati organizzati in 11 domini. Ogni evento segue il formato dominio:azione-in-kebab-case e possiede un payload TypeScript fortemente tipizzato.

Riferimento Completo degli Eventi

Esempio di Payload Tipizzato

Ogni evento possiede un payload TypeScript specifico. Ecco alcuni esempi rappresentativi per ciascun dominio principale:

// scrum:sprint-completed
{
  sprintId: string;
  name: string;
  velocity: number;
  completedStories: number;
  incompleteStories: number;
}

// time:entry-logged
{
  taskId: string;
  userId: string;
  minutes: number;
  date: string;       // ISO 8601
}

// cicd:build-failed
{
  pipelineId: string;
  error: string;
  stage: string;
  branch: string;
}

// economics:budget-alert
{
  project: string;
  percentUsed: number;
  threshold: number;   // es. 80
  remaining: number;
}

// code:dependency-alert
{
  package: string;
  severity: 'critical' | 'high' | 'medium' | 'low';
  advisory: string;
}
    

Pattern di Integrazione nei Server

L'integrazione dell'EventBus nei server segue un pattern standardizzato a 4 step che garantisce separazione delle responsabilità e opzionalita completa.

1. Creazione dell'EventBus (index.ts)

// servers/<nome>/src/index.ts
import { LocalEventBus } from '@mcp-suite/event-bus';

const eventBus = new LocalEventBus();
const suite = createMyServer(eventBus);
    

2. Iniezione nel Server (server.ts)

export function createMyServer(eventBus?: EventBus): McpSuiteServer {
  const suite = createMcpServer({
    name: 'my-server',
    version: '0.1.0',
    eventBus,
  });

  const store = new MyStore();

  // Tool che pubblica eventi
  registerCreateItem(suite.server, store, suite.eventBus);

  // Tool read-only: non serve l'eventBus
  registerListItems(suite.server, store);

  // Collaboration handlers
  if (suite.eventBus) {
    setupCollaborationHandlers(suite.eventBus, store);
  }

  return suite;
}
    

3. Pubblicazione negli Handler dei Tool (tools/*.ts)

export function registerCreateItem(
  server: McpServer,
  store: MyStore,
  eventBus?: EventBus,
): void {
  server.tool('create-item', 'Create a new item', schema, async (args) => {
    const item = store.create(args);

    // Fire-and-forget: eventBus può essere undefined
    eventBus?.publish('domain:item-created', {
      itemId: item.id,
      // ...payload tipizzato
    });

    return { content: [{ type: 'text', text: JSON.stringify(item) }] };
  });
}
    

4. Sottoscrizione (collaboration.ts)

export function setupCollaborationHandlers(
  eventBus: EventBus,
  store: MyStore,
): void {
  eventBus.subscribe('other-domain:event', (payload) => {
    // Reagisci all'evento
    store.updateSomething(payload);
  });
}
    

Pattern Matching con micromatch

Il metodo subscribePattern permette di sottoscrivere a gruppi di eventi usando pattern glob, grazie alla libreria micromatch:

// Tutti gli eventi del dominio scrum
eventBus.subscribePattern('scrum:*', (event, payload) => {
  console.log(`Scrum event: ${event}`, payload);
});

// Tutti gli eventi di completamento
eventBus.subscribePattern('*:*-completed', (event, payload) => {
  console.log(`Completed: ${event}`, payload);
});

// Workflow orchestrator: intercetta TUTTI gli eventi
eventBus.subscribePattern('*', async (event, payload) => {
  const workflows = store.getActiveWorkflowsByTrigger(event);
  for (const workflow of workflows) {
    if (matchesConditions(workflow.triggerConditions, payload)) {
      await executeWorkflow(workflow, payload);
    }
  }
});
    

Matrice di Collaborazione: Publisher e Subscriber

La matrice di collaborazione documenta tutti i flussi di eventi tra i 22 server della suite. Ogni server e classificato in base al suo ruolo nella rete di eventi.

Chi Sottoscrive Cosa

Classificazione dei Server

I 22 server della suite si classificano in quattro categorie in base al loro ruolo nella rete di collaborazione:

Server Bidirezionali (Pubblicano + Sottoscrivono)

Questi 5 server partecipano attivamente alla rete di eventi in entrambe le direzioni:

Server Solo Publisher (7 server)

Generano eventi ma non reagiscono a eventi esterni:

• cicd-monitor (2 eventi): hub per eventi DevOps
• code-review (2 eventi): notifiche analisi codice
• dependency-manager (1 evento): alert vulnerabilità
• db-schema-explorer (1 evento): suggerimenti indici
• test-generator (2 eventi): notifiche generazione test
• api-documentation (2 eventi): aggiornamenti docs
• performance-profiler (2 eventi): report performance

Server Solo Subscriber (1 server)

agile-metrics (3 eventi in ingresso): calcola metriche da eventi scrum senza pubblicare eventi propri direttamente dai tool.

Server Pass-Through (9 server)

Questi server accettano l'EventBus nel factory ma attualmente non pubblicano ne sottoscrivono. Sono predisposti per espansioni future:

• docker-compose, log-analyzer, data-mock-generator
• codebase-knowledge, regex-builder, http-client
• snippet-manager, project-scaffolding, environment-manager

ClientManager: Chiamate RPC Sincrone tra Server

Il ClientManager e il secondo meccanismo di comunicazione inter-server. A differenza dell'EventBus (fire-and-forget asincrono), il ClientManager consente a un server MCP di chiamare un tool su un altro server e ottenere il risultato in modo sincrono (request/response).

Server Chiamante          Client Manager          Server Target
      |                        |                       |
      |-- callTool('target',   |                       |
      |   'tool') ------------>|                       |
      |                        |-- MCP JSON-RPC ------>|
      |                        |                       |
      |                        |<-- JSON-RPC response--|
      |<-- risultato ----------|                       |
      |                        |                       |
      |  (sincrono: il chiamante attende il risultato) |
    

Regola pratica: usa l'EventBus quando vuoi notificare altri server, usa il ClientManager quando hai bisogno di dati da un altro server.

Esempio di Chiamata cross-server con ClientManager

export function registerForecastBudget(
  server: McpServer,
  store: EconomicsStore,
  clientManager?: McpClientManager,
): void {
  server.tool(
    'forecast-budget',
    'Prevedi quando il budget si esaurira',
    {
      projectName: z.string(),
      includeTimeData: z.boolean().optional().default(false),
    },
    async ({ projectName, includeTimeData }) => {
      const forecast = store.forecastBudget(projectName);
      const result: Record<string, unknown> = { ...forecast };

      // Cross-server call: recupera dati dal time-tracking
      if (includeTimeData && clientManager) {
        const timeResult = await clientManager.callTool(
          'time-tracking',     // Server target
          'get-timesheet',     // Tool da invocare
          {},                  // Argomenti
        );

        const content = (timeResult as {
          content: Array<{ type: string; text: string }>;
        }).content;
        const timesheet = JSON.parse(content[0].text);

        result.laborAnalysis = {
          trackedHours: timesheet.totalMinutes / 60,
          estimatedLaborCost: (timesheet.totalMinutes / 60) * 50,
        };
      }

      return {
        content: [{ type: 'text', text: JSON.stringify(result, null, 2) }],
      };
    },
  );
}
    

I 6 Scenari di Wiring Implementati

MCP Suite implementa 6 scenari di comunicazione cross-server che coinvolgono 5 server chiamanti e 3 server target:

Graceful Degradation

Il principio fondamentale del wiring e la graceful degradation: ogni tool funziona perfettamente anche senza ClientManager. Il wiring arricchisce il risultato, ma non e mai obbligatorio:

// Il pattern if (param && clientManager) garantisce:
// 1. Senza clientManager: il tool funziona normalmente
// 2. Senza il parametro trigger: il wiring non viene attivato
// 3. Con entrambi: il risultato viene arricchito con dati esterni
if (enrichFromExternal && clientManager) {
  // cross-server call
}
    

Flussi End-to-End

La vera potenza della collaborazione emerge nei flussi end-to-end che attraversano più server. Questi flussi nascono dalla composizione naturale degli eventi, senza orchestrazione centralizzata.

Flusso 1: Ciclo di Vita Sprint

[Creazione Sprint]
scrum-board:create-sprint
    |
    +-- scrum:sprint-started -----> (informativo, espansione futura)

[Lavoro durante lo Sprint]
scrum-board:update-task-status
    |
    +-- scrum:task-updated -------> agile-metrics  (cycle time)
    |                           +-> time-tracking  (auto-timer futuro)
    |                           +-> standup-notes  (contesto standup)

[Chiusura Sprint]
scrum-board:close-sprint
    |
    +-- scrum:sprint-completed ---> agile-metrics       (velocity)
    |                           +-> project-economics   (snapshot costo)
    |                           +-> retrospective-mgr   (auto-crea retro)
    |
    +-- scrum:story-completed ----> agile-metrics       (throughput)
    

Flusso 2: Tracking Tempo e Costi

[Registrazione Tempo]
time-tracking:log-time / stop-timer
    |
    +-- time:entry-logged --------> project-economics  (conversione costo)

[Registrazione Costo Diretto]
project-economics:log-cost
    |
    +-- economics:cost-updated ---> (informativo)

[Verifica Budget]
project-economics:get-budget-status
    |
    +-- economics:budget-alert ---> (notifica se >80%)
    

Flusso 3: Ciclo DevOps

[Monitoraggio Pipeline]
cicd-monitor:get-pipeline-status
    |
    +-- cicd:pipeline-completed --> (informativo)
    |
    +-- cicd:build-failed -------> retrospective-mgr  (item discussione)
                                +-> standup-notes      (blocker)
    

Flusso 4: Quality e Review

[Analisi Codice]
code-review:analyze-diff
    |
    +-- code:commit-analyzed -----> (informativo)

code-review:suggest-improvements
    |
    +-- code:review-completed ----> (informativo, codebase-knowledge futuro)

[Vulnerabilità]
dependency-manager:check-vulnerabilities
    |
    +-- code:dependency-alert ----> (informativo)
    

Flusso 5: Retrospettiva e Miglioramento Continuo

[Sprint Completato]
scrum:sprint-completed -----------> retrospective-mgr (auto-crea retro)

[Generazione Action Items]
retrospective-mgr:generate-action-items
    |
    +-- retro:action-item-created -> scrum-board (auto-crea task futuro)
    

Testing dell'EventBus e del Wiring

Per testare la collaborazione, Tech-MCP fornisce un MockEventBus che registra gli eventi pubblicati senza effetti collaterali:

export class MockEventBus implements EventBus {
  public published: Array<{ event: string; payload: unknown }> = [];

  async publish(event: string, payload: unknown): Promise<void> {
    this.published.push({ event, payload });
  }

  subscribe(): () => void { return () => {}; }
  subscribePattern(): () => void { return () => {}; }
  clear(): void { this.published = []; }

  // Utility per asserzioni
  wasPublished(eventName: string): boolean {
    return this.published.some((e) => e.event === eventName);
  }
}

// Uso nei test
it('should publish time:entry-logged event', async () => {
  const eventBus = new MockEventBus();
  const { server } = createTimeTrackingServer({
    eventBus,
    storeOptions: { inMemory: true },
  });
  const harness = await createTestHarness(server);

  await harness.client.callTool({
    name: 'log-time',
    arguments: { taskId: 'TASK-1', durationMinutes: 60 },
  });

  expect(eventBus.wasPublished('time:entry-logged')).toBe(true);
});
    

Per il wiring cross-server, i test usano InMemoryTransport per connettere i server in-process senza rete. L'ordine e critico: il server target deve connettersi al transport prima del client, perchè il client invia immediatamente il messaggio initialize al momento della connessione.

Sviluppi Futuri

La rete di collaborazione e progettata per crescere. Tra gli sviluppi pianificati:

• Redis EventBus: per deployment multi-processo, permettendo la collaborazione tra server in processi separati
• Nuovi eventi: scrum:backlog-prioritized, time:timer-started/time:timer-stopped, deploy:release-created
• Implementazione completa dei collaboration handler: molti handler contengono attualmente placeholder in attesa di implementazione
• Workflow orchestrator: un server che usa subscribePattern('*') per coordinare flussi complessi multi-step

Conclusioni

La collaborazione inter-server rappresenta il livello architetturale che trasforma una collezione di server indipendenti in un ecosistema integrato. L'EventBus fornisce il disaccoppiamento per notifiche e reazioni automatiche, mentre il ClientManager abilita l'arricchimento dei dati attraverso query sincrone cross-server.

I 29 eventi tipizzati, organizzati in 11 domini, creano una rete di collaborazione dove ogni azione significativa può innescare reazioni automatiche in server multipli. La classificazione dei server in bidirezionali, publisher-only, subscriber-only e pass-through garantisce chiarezza architetturale e facilità l'espansione futura.

Nel prossimo articolo affronteremo il testing professionale e il deploy in produzione, analizzando come testare i server MCP con test harness, InMemoryTransport, e come prepararli per ambienti di produzione con configurazione, logging e monitoring.

Il codice completo della collaborazione inter-server e disponibile nel repository Tech-MCP su GitHub.