Hct MCP Signals

HCT-MCP信号

模型上下文协议的协调层

在多代理系统中表达紧急性、定时和同步。

 ](https://www.npmjs.com/package/@hct-mcp/signals)  

\[！注意\] 状态：早期公开预览\ HCT-MCP信号 目前在 公开预览规范和实现是功能性的，但仍在不断发展。我们正在启动该项目，以促进合作、收集反馈和征求意见。期待变化。

______________________________________________________________________

🧠 问题

MCP将代理连接到工具，但代理如何协调 _彼此_?

标准MCP消息(tasks/send)缺乏以下词汇：

• 🚨 紧迫性：“放下一切，立即处理！”
• ⏱️ 时机：“我需要在500毫秒内得到这个。”
• ✋ 审批：“暂停，直到有人签字。”
• 🔄 循环：“重试，直到质量>90%。”
• 🛑 同步“在这里等着，等大家都赶上。”

没有这些信号，开发人员就会构建脆弱的、临时的状态机。

💡 解决方案：谐波协调

HCT-MCP信号 扩展了协议，有7个音乐原语被证明可以在没有中央指挥的情况下协调复杂的合奏。

______________________________________________________________________

⚡ 建筑

HCT信号无缝嵌入到标准MCP JSON-RPC消息中。现有服务器忽略它们；启用的代理利用它们进行高保真协调。

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)
Tempo: Allegro
    
    M->>A: Activate(Priority: High)
    A->>A: Processing...
    
    A->>M: tasks/sendSubscribe (FERMATA)
    Note right of A: Condition: Quality >H: Request Approval
    H-->>M: Approve
    
    M->>O: tasks/complete (Result + TACET)

______________________________________________________________________

🚀 安装

______________________________________________________________________

💻 快速开始

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

锈

use hct_mcp_signals::{cue, Tempo};

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

______________________________________________________________________

🔌 框架集成

HCT信号与框架无关。以下是它们如何增强各种代理架构：

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"

船员AI

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

谷歌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绞线/基岩

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

______________________________________________________________________

📜 完整规格

完整的协议规范可在 RFC.md.

🔗 规范来源

信号定义由自动生成 hct规范:

• spec.py, spec.ts, spec.go, spec.rs 通过CI同步

相关:

• hct-a2a -A2A扩展

🤝 贡献

请看 贡献.md.

______________________________________________________________________

📄 引用

如果您在研究或项目中使用HCT-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}
}

有关基本理论，请参阅 HCT纸.

______________________________________________________________________

许可证

Apache许可证2.0