多智能体协调协议

sequenceDiagram
    autonumber
    participant O as Orchestrator
    participant M as MCP Server
    participant A as Agent (Analyst)
    participant H as Human
    
    O->>M: tasks/send (CUE: "Analyze Q4")
    Note right of O: Urgency: 8 (High)<br/>Tempo: Allegro
    
    M->>A: Activate(Priority: High)
    A->>A: Processing...
    
    A->>M: tasks/sendSubscribe (FERMATA)
    Note right of A: Condition: Quality < 0.9<br/>Reason: "Ambiguous Data"
    
    M->>H: Request Approval
    H-->>M: Approve
    
    M->>O: tasks/complete (Result + TACET)

from hct_mcp_signals import cue, fermata

# 1. Dispatch with Urgency
signal = cue("orch", ["analyst"], urgency=9, tempo="presto")
mcp_client.send_tool_use("analyze", hct_signal=signal.to_mcp())

# 2. Hold for Approval
hold = fermata("analyst", "Needs Review", hold_type="human")

import { cue, Tempo } from '@hct-mcp/signals';

// 1. Dispatch with Urgency
const signal = cue({
    source: 'orch',
    targets: ['analyst'],
    urgency: 9,
    tempo: Tempo.PRESTO
});

use hct_mcp_signals::{cue, Tempo};

// 1. Builder Pattern
let signal = cue("orch", ["analyst"])
    .with_urgency(9)
    .with_tempo(Tempo::Presto)
    .build();

from langgraph.graph import StateGraph
from hct_mcp_signals import fermata, caesura

def router(state):
    signal = state.get("hct_signal", {})
    if signal.get("type") == "fermata":
        return "human_review"
    elif signal.get("type") == "caesura":
        return "end"
    return "continue"

from crewai import Task
from hct_mcp_signals import cue, vamp

# Embed HCT signal in task context for quality gating
Task(
    description="Analyze Q4 Trends",
    expected_output="Financial Report",
    context={"hct_signal": vamp("verifier", "confidence >= 0.9").to_mcp()}
)

from hct_mcp_signals import downbeat

# Sync point before parallel work
sync = downbeat("coordinator", "phase_2_start")
assistant.send({"content": "Starting phase 2", "hct_signal": sync.to_mcp()})

class CoordinatedAgent(Agent):
    async def on_message(self, message):
        signal = message.get("hct_signal")
        if signal and signal["type"] == "caesura":
            await self.emergency_shutdown(signal["payload"]["reason"])

class StrandsCoordinatedAgent(Agent):
    def handle_mcp_message(self, params):
        signal = params.get("hct_signal")
        urgency = signal.get("performance", {}).get("urgency", 5)
        
        if urgency >= 8:
            return self.priority_process(params)
        return self.normal_process(params)

class QualityControlledModule(dspy.Module):
    def forward(self, question):
        # Use VAMP signal for retry logic
        signal = vamp("dspy_module", "quality >= 0.9", quality_threshold=0.9)
        # Emit signal to observer...
        return self.generate(question)

@gateway.route
def route_with_urgency(request):
    signal = request.get("hct_signal", {})
    urgency = signal.get("performance", {}).get("urgency", 5)
    
    if urgency >= 9:
        return "fast_model"
    elif signal.get("type") == "fermata":
        return "careful_model"
    return "default_model"

class MemoryCoordinatedAgent(Agent):
    def hibernate(self, duration_ms):
        # Signal agent is going inactive (TACET)
        signal = tacet(self.name, duration_ms=duration_ms)
        self.broadcast(signal.to_mcp())

@software{wiest2025hctsignals,
  title = {HCT-MCP Signals: The Coordination Layer for Model Context Protocol},
  author = {Wiest, Stefan},
  year = {2025},
  url = {https://github.com/stefanwiest/hct-mcp-signals},
  version = {0.8.0}
}