feat(mcp): echter Push an Agenten über MCP-Notifications — ohne blockierenden Tool-Call
DIE WURZEL, endlich an der Wurzel gefasst. Bisher konnte ein Agent nur
empfangen, WÄHREND er in `agenthub_work` blockierte. Kehrte der Aufruf leer
zurück und das Modell startete ihn nicht neu, war der Agent unerreichbar —
ein Mensch musste ihn anstoßen. Wir haben einen ganzen Tag um diese Tatsache
herum gebaut (Watchdog, Check-in-Kanal, längere Timeouts, notListening) statt
sie aufzulösen. Der CEO hat zu Recht darauf bestanden, das Protokoll selbst
zu nutzen.
MCP kann Server→Client jederzeit `notifications/message` schicken. Die Bridge
ist ohnehin ein langlebiger Prozess pro Agent: sie hängt sich jetzt dauerhaft
an den SSE-Stream des Hubs (`pushChannel.ts`) und reicht relevante Ereignisse
als Notification hoch — neue/zurückgegebene Task, Nachricht, beantwortete
Frage, Abbruch. Bindung erfolgt, sobald der Agent seinen Namen nennt
(hello/work), inklusive Alias-Auflösung; Reconnect mit Backoff.
⚠️ Die entscheidende Falle: `sendLoggingMessage` prüft `_capabilities.logging`
und verwirft die Notification sonst STILL. Der Server deklarierte gar keine
Capabilities — daran wäre dieser Weg unbemerkt gescheitert. Jetzt deklariert.
Live verifiziert mit einem echten MCP-Client: Nachricht am Hub angelegt →
Notification kam beim Client an, ohne dass dieser in einem Tool-Call wartete.
OFFEN und bewusst nicht behauptet: ob die CLI-Hosts (Codex/Kimi) die
Notification dem Modell zeigen — das ist client-abhängig und empirisch zu
prüfen. Deshalb bleiben Check-in-Kanal und Work-Loop als Netz bestehen.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
e5bd28488d
commit
e8c2e43cb0
122
src/mcp/pushChannel.ts
Normal file
122
src/mcp/pushChannel.ts
Normal file
@ -0,0 +1,122 @@
|
||||
import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
|
||||
import { parseSSEBuffer } from '../cli/commands/watch.js';
|
||||
import { remoteClient } from '../cli/remoteClient.js';
|
||||
|
||||
/**
|
||||
* Echter Push an den Agenten über MCP — ohne blockierenden Tool-Call.
|
||||
*
|
||||
* WARUM DAS HIER STEHT (die Lehre aus einem ganzen Tag):
|
||||
* Bisher konnte ein Agent nur empfangen, während er in `agenthub_work`
|
||||
* blockierte. Kehrte der Aufruf leer zurück und das Modell startete ihn nicht
|
||||
* neu, war der Agent unerreichbar — kein Reopen, keine Nachricht kam an, und
|
||||
* ein Mensch musste ihn anstoßen. Wir haben um diese Tatsache herum gebaut
|
||||
* (Watchdog, Check-in-Kanal, längere Timeouts) statt sie aufzulösen.
|
||||
*
|
||||
* MCP kann das von Haus aus: der Server darf dem Client JEDERZEIT
|
||||
* `notifications/message` schicken. Unsere Bridge ist ohnehin ein langlebiger
|
||||
* Prozess pro Agent — sie hängt sich dauerhaft an den SSE-Stream des Hubs und
|
||||
* reicht relevante Ereignisse als MCP-Notification hoch. Kein Blockieren,
|
||||
* kein Timeout, keine Loop-Disziplin nötig.
|
||||
*
|
||||
* ⚠️ VORAUSSETZUNG: Der Server MUSS die `logging`-Capability deklarieren —
|
||||
* `sendLoggingMessage` prüft `_capabilities.logging` und verwirft die
|
||||
* Notification sonst STILL. Genau daran wäre dieser Weg unbemerkt gescheitert.
|
||||
*
|
||||
* ⚠️ OFFEN (client-abhängig, nicht von uns entscheidbar): ob der jeweilige
|
||||
* CLI-Host die Notification dem Modell zeigt. Das ist empirisch zu prüfen und
|
||||
* darf nicht behauptet werden. Deshalb bleibt der Check-in-Kanal als Netz
|
||||
* bestehen — dieser Push ist die Verbesserung, nicht der alleinige Verlass.
|
||||
*/
|
||||
|
||||
/** Reconnect-Backoff, wenn der SSE-Stream abreißt (Hub-Neustart o. Ä.). */
|
||||
const RECONNECT_DELAYS_MS = [1000, 3000, 8000];
|
||||
|
||||
interface PushState {
|
||||
agent: string;
|
||||
names: Set<string>;
|
||||
stop: () => void;
|
||||
}
|
||||
|
||||
let active: PushState | undefined;
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
/** Betrifft dieses Event den Agenten, für den wir pushen? */
|
||||
function relevantFor(
|
||||
event: { type?: string; action?: string; assignedTo?: string; status?: string },
|
||||
names: Set<string>,
|
||||
): boolean {
|
||||
const target = String(event.assignedTo ?? '').toLowerCase();
|
||||
if (target && names.has(target)) return true;
|
||||
// Ohne Adressat ist ein Task-Event nur interessant, wenn es ein Reopen sein
|
||||
// könnte — der Claim wird dabei abgeräumt, der Adressat bleibt aber stehen,
|
||||
// also greift oben bereits `assignedTo`. Alles andere ignorieren wir bewusst:
|
||||
// Rauschen entwertet den Kanal (Lehre aus dem Watchdog-Spam).
|
||||
return false;
|
||||
}
|
||||
|
||||
function describe(event: { type?: string; action?: string; id?: string; status?: string }): string {
|
||||
const id = event.id ?? '?';
|
||||
if (event.type === 'message') return `Neue Nachricht ${id} für dich — agenthub_inbox lesen.`;
|
||||
if (event.type === 'ask') return `Deine Frage ${id} wurde beantwortet — agenthub_ask_list.`;
|
||||
if (event.status === 'open') return `${id} wartet auf dich (neu oder zurückgegeben) — agenthub_work aufrufen.`;
|
||||
if (event.status === 'cancelled') return `${id} wurde abgebrochen — STOPP, nicht weiterbauen.`;
|
||||
return `${id}: ${event.type}/${event.action} (${event.status ?? '—'}).`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Bindet die Bridge an einen Agenten und beginnt zu pushen. Idempotent:
|
||||
* derselbe Agent startet den Stream nicht doppelt, ein anderer löst den alten ab.
|
||||
*/
|
||||
export function bindPushChannel(server: McpServer, serverUrl: string, agent: string): void {
|
||||
const name = agent.trim().toLowerCase();
|
||||
if (!name || active?.agent === name) return;
|
||||
active?.stop();
|
||||
|
||||
let stopped = false;
|
||||
const state: PushState = { agent: name, names: new Set([name]), stop: () => { stopped = true; } };
|
||||
active = state;
|
||||
|
||||
void (async () => {
|
||||
// Aliase mitnehmen, damit ein umbenannter Agent seine Events weiter bekommt.
|
||||
try {
|
||||
const identity = await remoteClient.getAgentIdentity(serverUrl, agent);
|
||||
for (const alias of identity.names) state.names.add(alias.toLowerCase());
|
||||
} catch { /* Namensvergleich reicht */ }
|
||||
|
||||
let attempt = 0;
|
||||
while (!stopped) {
|
||||
try {
|
||||
const res = await fetch(new URL('/events', serverUrl).toString(), {
|
||||
headers: { Accept: 'text/event-stream' },
|
||||
});
|
||||
if (!res.body) throw new Error('SSE ohne Body');
|
||||
attempt = 0;
|
||||
const reader = res.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
while (!stopped) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
if (value) buffer += decoder.decode(value, { stream: true });
|
||||
const { events, remaining } = parseSSEBuffer(buffer);
|
||||
buffer = remaining;
|
||||
for (const event of events) {
|
||||
if (!relevantFor(event, state.names)) continue;
|
||||
try {
|
||||
await server.server.sendLoggingMessage({
|
||||
level: 'info',
|
||||
logger: 'agenthub',
|
||||
data: { agenthub: describe(event), event },
|
||||
});
|
||||
} catch { /* Client mag Notifications nicht — kein Grund abzubrechen */ }
|
||||
}
|
||||
}
|
||||
} catch { /* Hub weg oder Neustart — unten neu versuchen */ }
|
||||
if (stopped) return;
|
||||
await sleep(RECONNECT_DELAYS_MS[Math.min(attempt++, RECONNECT_DELAYS_MS.length - 1)]!);
|
||||
}
|
||||
})();
|
||||
}
|
||||
@ -19,6 +19,7 @@ import {
|
||||
import { createHandoff, getHandoff } from '../core/services/handoffService.js';
|
||||
import { appendTaskLog } from '../core/services/taskLogService.js';
|
||||
import { resolvePending, hasPending } from '../core/services/checkinService.js';
|
||||
import { bindPushChannel } from './pushChannel.js';
|
||||
import { createAsk, listAsks, answerAsk, escalateAsk } from '../core/services/askService.js';
|
||||
import type { Ask } from '../core/schema.js';
|
||||
import { createMessage, listInbox } from '../core/services/messageService.js';
|
||||
@ -200,12 +201,24 @@ export function waitForTask<T>(
|
||||
export async function startMcpServer(cwd: string): Promise<void> {
|
||||
const { root, serverUrl } = await resolveMcpContext(cwd);
|
||||
const remote = !!serverUrl;
|
||||
const server = new McpServer({ name: 'agenthub', version: VERSION });
|
||||
// `logging` MUSS deklariert werden: sendLoggingMessage prueft
|
||||
// _capabilities.logging und verwirft die Notification sonst kommentarlos.
|
||||
// Ohne diese Zeile gaebe es keinen Push — und man saehe nicht, warum.
|
||||
const server = new McpServer(
|
||||
{ name: 'agenthub', version: VERSION },
|
||||
{ capabilities: { logging: {} } },
|
||||
);
|
||||
|
||||
/** Bridge an den Agenten binden, sobald er sich zu erkennen gibt. */
|
||||
const bindPush = (agent?: string) => {
|
||||
if (remote && agent) bindPushChannel(server, serverUrl!, agent);
|
||||
};
|
||||
|
||||
server.tool('agenthub_hello', 'Announce yourself to the hub (presence). Run once when you start.',
|
||||
{ agent: z.string(), role: z.string().optional() },
|
||||
async ({ agent, role }) => {
|
||||
if (remote) await remoteClient.announce(serverUrl!, agent, role ?? 'implementer');
|
||||
bindPush(agent);
|
||||
return asText(`AgentHub: ${agent} joined (${role ?? 'implementer'})`);
|
||||
});
|
||||
|
||||
@ -213,6 +226,7 @@ export async function startMcpServer(cwd: string): Promise<void> {
|
||||
'Block until there is work for you, then return it. The default 50s timeout stays below known MCP client limits; SSE still wakes immediately, so this only causes more empty wake-ups and does not add delivery latency. IMPLEMENTER: waits for a task addressed to you (newly delegated OR reopened), claims it, returns it with its handoff (loop: work -> implement -> agenthub_task_review -> work). ARCHITECT/REVIEWER: waits for any task submitted to review and returns the pending set to approve (agenthub_task_done) or reject (agenthub_task_reopen) — so review submissions reach you in realtime.',
|
||||
{ agent: z.string(), role: z.string().optional(), timeoutSec: z.number().optional(), unattended: z.boolean().optional() },
|
||||
async ({ agent, role, timeoutSec, unattended }) => {
|
||||
bindPush(agent);
|
||||
const ctx: AgentContext = { serverUrl, projectCwd: root, agent, role: role ?? 'implementer' };
|
||||
if (remote) await remoteClient.setLoop(serverUrl!, agent, true);
|
||||
const leave = async (reason: string) => {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user