HCT-MCP Signals

[!NOTE] Status: Early Stage Public Preview
HCT-MCP Signals is currently in Public Preview. The specification and implementations are functional but are still evolving. We are opening the project to foster collaboration, gather feedback, and solicit contributions. Expect changes.

🧠 The Problem

MCP connects agents to tools, but how do agents coordinate with each other?

Standard MCP messages (tasks/send) lack the vocabulary for:

• 🚨 Urgency: "Drop everything and process this now!"
• ⏱️ Timing: "I need this in < 500ms."
• ✋ Approvals: "Pause until a human signs off."
• 🔄 Loops: "Retry until quality > 90%."
• 🛑 Sync: "Wait here until everyone catches up."

Without these signals, developers build brittle, ad-hoc state machines.

💡 The Solution: Harmonic Coordination

HCT-MCP Signals extends the protocol with 7 musical primitives proven to coordinate complex ensembles without a central conductor.

⚡ Architecture

HCT Signals embed seamlessly into standard MCP JSON-RPC messages. Existing servers ignore them; enabled agents leverage them for high-fidelity coordination.

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)

🚀 Installation

💻 Quick Start

Python

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")

TypeScript

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

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

Rust

use hct_mcp_signals::{cue, Tempo};

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

🔌 Framework Integrations

HCT Signals are framework-agnostic. Here is how they enhance various agent architectures:

LangGraph

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"

CrewAI

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()}
)

AutoGen

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()})

Google ADK

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"])

AWS Strands / Bedrock

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)

DSPy

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)

TensorZero

@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"

Letta (MemGPT)

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())

📜 Full Specification

The complete protocol specification is available in RFC.md.

🤝 Contributing

Please see CONTRIBUTING.md.

License

Apache License 2.0