mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
feat(mcp): forward MCP notifications/progress as live tool.progress events
MCP tool calls now send _meta.progressToken, so servers that emit
notifications/progress (Meshy refine, creative-lab) report live progress.
The executor binds a per-call relay; the MCP handler forwards each
notification to tool_progress_callback as a throttled (<=1 per 500 ms,
final always) tool.progress event. Surfaces:
- gateway event stream + legacy TUI gateway: tool.progress
{tool_id, name, progress, total, percent, message} (contract registered)
- Desktop: running tool row shows '42% · <message>' beside its timer;
plugins subscribe with ctx.onEvent('tool.progress')
- CLI: the running tool spinner shows the same text
Documented in the Desktop plugin SDK guide.
This commit is contained in:
1 parent
601491feb3
commit
60affe7fe2
19 files changed
+446
-10
No files matched your search
@@ -749,7 +749,9 @@ def _dispatch_authorized_once(
|
||||
from agent.terminal_approval_batch import prepare_current_terminal
|
||||
prepare_current_terminal(ref)
|
||||
_advance_start_order(lambda: _begin_tool_execution(agent, ref, display_index))
|
||||
return _run_with_activity_heartbeat(agent, ref.name, lambda: execute(ref.args))
|
||||
from agent.tool_progress_relay import bound_tool_progress
|
||||
with bound_tool_progress(agent.tool_progress_callback, ref.name, ref.call_id):
|
||||
return _run_with_activity_heartbeat(agent, ref.name, lambda: execute(ref.args))
|
||||
|
||||
|
||||
def _run_agent_tool_execution_middleware(
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Live progress from inside a running tool call (MCP ``notifications/progress``) onto the agent's
|
||||
``tool_progress_callback`` as ``tool.progress`` events.
|
||||
|
||||
The executor binds a relay for each dispatched call (``bound_tool_progress``); a tool reads it on its
|
||||
own thread with ``current_tool_progress()`` and may invoke the result from any thread (the MCP client
|
||||
fires it on the MCP event loop). Emission is throttled per call so a chatty server cannot flood the
|
||||
UI; the first update and the one reaching ``total`` always pass.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import contextvars
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Iterator, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TOOL_PROGRESS_EVENT = "tool.progress"
|
||||
_MIN_INTERVAL_S = 0.5
|
||||
|
||||
ProgressReporter = Callable[[float, Optional[float], Optional[str]], None]
|
||||
_current: contextvars.ContextVar[Optional[ProgressReporter]] = contextvars.ContextVar("tool_progress_reporter", default=None)
|
||||
|
||||
|
||||
def _reporter(callback, tool_name: str, tool_call_id: str) -> ProgressReporter:
|
||||
lock = threading.Lock()
|
||||
last = [0.0]
|
||||
|
||||
def report(progress: float, total: Optional[float] = None, message: Optional[str] = None) -> None:
|
||||
done = total is not None and total > 0 and progress >= total
|
||||
now = time.monotonic()
|
||||
with lock:
|
||||
if last[0] and not done and now - last[0] < _MIN_INTERVAL_S:
|
||||
return
|
||||
last[0] = now
|
||||
percent = round(min(max(progress / total, 0.0), 1.0) * 100) if total else None
|
||||
try:
|
||||
callback(TOOL_PROGRESS_EVENT, tool_name, message or None, None, tool_call_id=tool_call_id,
|
||||
progress=float(progress), total=None if total is None else float(total), percent=percent)
|
||||
except Exception: # health: allow BLE001 -- a UI sink failing must never fail the tool call
|
||||
logger.debug("tool.progress callback failed for %s", tool_name, exc_info=True)
|
||||
return report
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def bound_tool_progress(callback, tool_name: str, tool_call_id: str) -> Iterator[None]:
|
||||
"""Bind this call's relay for the block (no-op when the agent has no progress callback)."""
|
||||
token = _current.set(_reporter(callback, tool_name, tool_call_id or "") if callback else None)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_current.reset(token)
|
||||
|
||||
|
||||
def current_tool_progress() -> Optional[ProgressReporter]:
|
||||
"""The running call's reporter, or None outside a dispatched tool call."""
|
||||
return _current.get()
|
||||
|
||||
|
||||
def tool_progress_payload(name: Optional[str], message: Optional[str], kwargs: dict) -> dict:
|
||||
"""The ``tool.progress`` wire payload every viewer surface publishes (gateway event stream and
|
||||
the legacy TUI gateway): ``percent`` is 0-100 when the server sent a ``total``, else None."""
|
||||
return {"tool_id": str(kwargs.get("tool_call_id") or ""), "name": str(name or ""),
|
||||
"progress": float(kwargs.get("progress") or 0.0), "total": kwargs.get("total"),
|
||||
"percent": kwargs.get("percent"), "message": str(message) if message else None}
|
||||
@@ -27,7 +27,7 @@ import { handleOnboardingEvent } from './onboarding'
|
||||
import { handleControlEvent } from './session-control'
|
||||
import { handleSessionInfoEvent } from './session-info'
|
||||
import { handleStatusEvent } from './status'
|
||||
import { handleToolEvent } from './tools'
|
||||
import { handleToolEvent, handleToolProgressEvent } from './tools'
|
||||
import type { GatewayEventContext, GatewayEventDeps, GatewayEventHandler } from './types'
|
||||
|
||||
export type { GatewayEventDeps } from './types'
|
||||
@@ -91,6 +91,7 @@ const HANDLERS: GatewayEventHandler[] = [
|
||||
handleDeferredPickEvent,
|
||||
handleControlEvent,
|
||||
handleMessageStreamEvent,
|
||||
handleToolProgressEvent,
|
||||
handleToolEvent,
|
||||
handleInputRequestEvent,
|
||||
handleDesktopBridgeEvent,
|
||||
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
import type { GatewayEvent } from '@hermes/shared'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
|
||||
import { $toolProgress, $toolProgressFor, formatToolProgress } from '@/store/tool-progress'
|
||||
|
||||
import { handleToolProgressEvent } from './tools'
|
||||
import type { GatewayEventContext } from './types'
|
||||
|
||||
function deliver(event: GatewayEvent) {
|
||||
return handleToolProgressEvent({
|
||||
event,
|
||||
payload: event.payload,
|
||||
sessionId: 's1',
|
||||
occurredAt: 1,
|
||||
isActiveEvent: true,
|
||||
deps: {
|
||||
flushQueuedDeltas: vi.fn(),
|
||||
updateSessionState: vi.fn(),
|
||||
sessionInterrupted: () => false,
|
||||
upsertToolCall: vi.fn(),
|
||||
nativeSubagentSessionsRef: { current: new Set() }
|
||||
}
|
||||
} as unknown as GatewayEventContext)
|
||||
}
|
||||
|
||||
afterEach(() => $toolProgress.set({}))
|
||||
|
||||
it('shows tool.progress on its running row and drops it when the call completes', () => {
|
||||
const progress = {
|
||||
type: 'tool.progress',
|
||||
session_id: 's1',
|
||||
payload: { tool_id: 'c1', name: 'mcp__meshy__refine', progress: 42, total: 100, percent: 42, message: 'refining' }
|
||||
} as unknown as GatewayEvent
|
||||
|
||||
expect(deliver(progress)).toBe(true)
|
||||
// MCP rows render each batched call as `<call id>:<index>`.
|
||||
expect(formatToolProgress($toolProgressFor('c1:0').get())).toBe('42% · refining')
|
||||
|
||||
deliver({ type: 'tool.complete', session_id: 's1', payload: { tool_id: 'c1', name: 'mcp__meshy__refine' } } as GatewayEvent)
|
||||
expect($toolProgressFor('c1:0').get()).toBeUndefined()
|
||||
})
|
||||
@@ -15,6 +15,7 @@ import { reportMcpToolResult } from '@/store/suggestion-providers/repair'
|
||||
import { invalidateSkillSuggestionIndex } from '@/store/suggestion-providers/skill'
|
||||
import { restoreSessionTodosFromSnapshot } from '@/store/todos'
|
||||
import { setSessionDraftingTool } from '@/store/tool-drafting'
|
||||
import { clearToolProgress, setToolProgress } from '@/store/tool-progress'
|
||||
import { notifyWorkspaceChanged, toolChangedPath, toolMayMutateFiles } from '@/store/workspace-events'
|
||||
|
||||
import { SUBAGENT_EVENT_TYPES, toTodoPayload } from '../utils'
|
||||
@@ -69,6 +70,31 @@ function appliedLayout(name: string | undefined, result: ToolCallMessagePart['re
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** tool.progress (live MCP progress for a running call): a row annotation, never a transcript
|
||||
* mutation. Also retires it on tool.complete, which then falls through to handleToolEvent. */
|
||||
export function handleToolProgressEvent({ deps, event, payload, sessionId }: GatewayEventContext): boolean {
|
||||
const toolId = typeof payload?.tool_id === 'string' ? payload.tool_id : ''
|
||||
|
||||
if (event.type === 'tool.complete') {
|
||||
clearToolProgress(toolId)
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
if (event.type !== 'tool.progress') {
|
||||
return false
|
||||
}
|
||||
|
||||
if (sessionId && !deps.sessionInterrupted(sessionId)) {
|
||||
setToolProgress(toolId, {
|
||||
message: typeof payload?.message === 'string' && payload.message ? payload.message : null,
|
||||
percent: typeof payload?.percent === 'number' ? payload.percent : null
|
||||
})
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
/** tool.generating / tool.start / tool.complete / subagent.*. */
|
||||
export function handleToolEvent(ctx: GatewayEventContext): boolean {
|
||||
const { deps, event, payload, sessionId, isActiveEvent, occurredAt } = ctx
|
||||
|
||||
@@ -52,6 +52,7 @@ import { recordPreviewArtifact } from '@/store/preview-status'
|
||||
import { sessionApprovalRequest } from '@/store/prompts'
|
||||
import { $showToolActivity } from '@/store/tool-activity'
|
||||
import { $toolRowDismissed, dismissToolRow } from '@/store/tool-dismiss'
|
||||
import { $toolProgressFor, formatToolProgress } from '@/store/tool-progress'
|
||||
import {
|
||||
$anyToolDisclosureOpen,
|
||||
$hideCodeDiffs,
|
||||
@@ -344,6 +345,22 @@ function useDisclosureOpen(disclosureId: string, fallbackOpen = false): boolean
|
||||
return persistedOpen ?? fallbackOpen
|
||||
}
|
||||
|
||||
/** A running row's trailing meta: live MCP progress (`42% · message`, when the server sends it) + the timer. */
|
||||
function ToolRunningMeta({ seconds, toolCallId = '' }: { seconds: number; toolCallId?: string }) {
|
||||
const progressText = formatToolProgress(useStore($toolProgressFor(toolCallId)))
|
||||
|
||||
return (
|
||||
<>
|
||||
{progressText && (
|
||||
<span className={cn(SCAFFOLD_META_CLASS, 'max-w-64 truncate')} data-slot="tool-progress" title={progressText}>
|
||||
{progressText}
|
||||
</span>
|
||||
)}
|
||||
<ActivityTimerText className={SCAFFOLD_META_CLASS} seconds={seconds} />
|
||||
</>
|
||||
)
|
||||
}
|
||||
|
||||
function ToolEntry({ part }: ToolEntryProps) {
|
||||
const { t } = useI18n()
|
||||
const copy = t.assistant.tool
|
||||
@@ -541,7 +558,7 @@ function ToolEntry({ part }: ToolEntryProps) {
|
||||
const trailing = !embedded ? (
|
||||
<span className="flex shrink-0 items-center gap-1.5">
|
||||
<TimelineTimestamp className={SCAFFOLD_META_CLASS} completedAt={completedAt} timestamp={timestamp} />
|
||||
{isPending && <ActivityTimerText className={SCAFFOLD_META_CLASS} seconds={elapsed} />}
|
||||
{isPending && <ToolRunningMeta seconds={elapsed} toolCallId={toolCallId} />}
|
||||
</span>
|
||||
) : undefined
|
||||
|
||||
|
||||
@@ -87,6 +87,8 @@ export type GatewayEventPayload = {
|
||||
name?: string
|
||||
tool_id?: string
|
||||
tool_call_id?: string
|
||||
/** tool.progress: 0-100 when the MCP server sent a total. */
|
||||
percent?: number | null
|
||||
args?: unknown
|
||||
arguments?: unknown
|
||||
context?: string
|
||||
|
||||
@@ -39,6 +39,7 @@ export const UNSCOPED_STREAM_EVENT_TYPES = new Set([
|
||||
'thinking.delta',
|
||||
'tool.complete',
|
||||
'tool.generating',
|
||||
'tool.progress',
|
||||
'tool.start',
|
||||
'vault.code.expire',
|
||||
'vault.code.request',
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import { atom, computed, type ReadableAtom } from 'nanostores'
|
||||
|
||||
/** Latest live progress of a running tool call (`tool.progress`, from MCP `notifications/progress`). */
|
||||
export interface ToolProgress {
|
||||
message: null | string
|
||||
percent: null | number
|
||||
}
|
||||
|
||||
// Keyed by tool call id. Ephemeral by design: progress only exists while a call
|
||||
// runs, so it is never persisted and the row drops it once the call completes.
|
||||
export const $toolProgress = atom<Record<string, ToolProgress>>({})
|
||||
|
||||
const progressCache = new Map<string, ReadableAtom<ToolProgress | undefined>>()
|
||||
|
||||
export function $toolProgressFor(toolCallId: string): ReadableAtom<ToolProgress | undefined> {
|
||||
let cached = progressCache.get(toolCallId)
|
||||
|
||||
if (!cached) {
|
||||
// MCP rows render per batched call as `<call id>:<index>`; progress is reported for the call.
|
||||
const callId = toolCallId.replace(/:\d+$/, '')
|
||||
cached = computed($toolProgress, rows => rows[toolCallId] ?? rows[callId])
|
||||
progressCache.set(toolCallId, cached)
|
||||
}
|
||||
|
||||
return cached
|
||||
}
|
||||
|
||||
export function setToolProgress(toolCallId: string, progress: ToolProgress): void {
|
||||
if (toolCallId) {
|
||||
$toolProgress.set({ ...$toolProgress.get(), [toolCallId]: progress })
|
||||
}
|
||||
}
|
||||
|
||||
export function clearToolProgress(toolCallId: string): void {
|
||||
const rows = $toolProgress.get()
|
||||
|
||||
if (toolCallId && rows[toolCallId]) {
|
||||
const { [toolCallId]: _done, ...rest } = rows
|
||||
$toolProgress.set(rest)
|
||||
}
|
||||
}
|
||||
|
||||
/** `42% · refining mesh` — the text a running tool row shows beside its timer. */
|
||||
export function formatToolProgress(progress: ToolProgress | undefined): string {
|
||||
if (!progress) {
|
||||
return ''
|
||||
}
|
||||
|
||||
return [progress.percent === null ? '' : `${progress.percent}%`, progress.message ?? ''].filter(Boolean).join(' · ')
|
||||
}
|
||||
@@ -5748,6 +5748,15 @@ export interface ToolOutputRiskPayload {
|
||||
findings: string[]
|
||||
redacted: boolean
|
||||
}
|
||||
/** ``agent.tool_progress_relay.tool_progress_payload`` — an MCP server's ``notifications/progress`` for a running call, throttled to one per 500 ms per call. */
|
||||
export interface ToolProgressPayload {
|
||||
tool_id: string
|
||||
name: string
|
||||
progress: number
|
||||
total?: number | null
|
||||
percent?: number | null
|
||||
message?: string | null
|
||||
}
|
||||
/** ``tool_progress._normalize_todo_state`` — full task snapshot. */
|
||||
export interface TodoUpdatedPayload {
|
||||
todos: unknown[]
|
||||
@@ -7081,6 +7090,8 @@ export interface BackendGatewayEventMap {
|
||||
'tool.generating': ToolGeneratingPayload
|
||||
/** Tool output was classified as risky (prompt-injection / secret findings). */
|
||||
'tool.output_risk': ToolOutputRiskPayload
|
||||
/** Live progress of a running tool call (MCP notifications/progress). */
|
||||
'tool.progress': ToolProgressPayload
|
||||
/** A tool call began (stable id + full args). */
|
||||
'tool.start': ToolStartPayload
|
||||
/** Barge-in: the spoken interjection interrupted the turn; no payload. */
|
||||
@@ -7169,6 +7180,7 @@ export const GATEWAY_EVENT_TYPES = [
|
||||
'tool.complete',
|
||||
'tool.generating',
|
||||
'tool.output_risk',
|
||||
'tool.progress',
|
||||
'tool.start',
|
||||
'voice.interrupted',
|
||||
'voice.partial',
|
||||
|
||||
@@ -43809,6 +43809,67 @@
|
||||
"title": "ToolOutputRiskPayload",
|
||||
"type": "object"
|
||||
},
|
||||
"ToolProgressPayload": {
|
||||
"additionalProperties": false,
|
||||
"description": "``agent.tool_progress_relay.tool_progress_payload`` \u2014 an MCP server's\n``notifications/progress`` for a running call, throttled to one per 500 ms per call.",
|
||||
"properties": {
|
||||
"tool_id": {
|
||||
"title": "Tool Id",
|
||||
"type": "string"
|
||||
},
|
||||
"name": {
|
||||
"title": "Name",
|
||||
"type": "string"
|
||||
},
|
||||
"progress": {
|
||||
"title": "Progress",
|
||||
"type": "number"
|
||||
},
|
||||
"total": {
|
||||
"anyOf": [
|
||||
{
|
||||
"type": "number"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
],
|
||||
"default": null,
|
||||
"title": "Total"
|
||||
},
|
||||
"percent": {
|
||||
"anyOf": [
|
||||
{
|
||||
"type": "integer"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
],
|
||||
"default": null,
|
||||
"title": "Percent"
|
||||
},
|
||||
"message": {
|
||||
"anyOf": [
|
||||
{
|
||||
"type": "string"
|
||||
},
|
||||
{
|
||||
"type": "null"
|
||||
}
|
||||
],
|
||||
"default": null,
|
||||
"title": "Message"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"tool_id",
|
||||
"name",
|
||||
"progress"
|
||||
],
|
||||
"title": "ToolProgressPayload",
|
||||
"type": "object"
|
||||
},
|
||||
"ToolShowRow": {
|
||||
"additionalProperties": false,
|
||||
"properties": {
|
||||
@@ -49177,6 +49238,18 @@
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "tool.progress",
|
||||
"summary": "Live progress of a running tool call (MCP notifications/progress).",
|
||||
"params": [
|
||||
{
|
||||
"name": "payload",
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ToolProgressPayload"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "tool.start",
|
||||
"summary": "A tool call began (stable id + full args).",
|
||||
|
||||
@@ -135,6 +135,10 @@ def publish_progress(turn, event_type, name, preview, kwargs) -> None:
|
||||
_publish_subagent(turn, event_type, _subagent_payload(
|
||||
event_type, name, preview, kwargs, show_reasoning=show_reasoning(turn)))
|
||||
return
|
||||
if event_type == "tool.progress":
|
||||
from agent.tool_progress_relay import tool_progress_payload
|
||||
turn._publish_execution(event_type, tool_progress_payload(name, preview, kwargs))
|
||||
return
|
||||
if event_type != "reasoning.available" and event_type not in _MOA_KEYS:
|
||||
return
|
||||
from gateway.session_cli_turn import classic_cli_turn
|
||||
|
||||
@@ -763,6 +763,7 @@ class CLIStreamMixin:
|
||||
|
||||
if event_type == "tool.completed":
|
||||
self._tool_start_time = 0.0
|
||||
self.__dict__.pop("_tool_spinner_base", None)
|
||||
self._turn_summary_record(
|
||||
function_name, kwargs.get("result"), kwargs.get("is_error", False))
|
||||
# Focus view: count the hidden scrollback line for the post-turn recovery report.
|
||||
@@ -810,20 +811,28 @@ class CLIStreamMixin:
|
||||
self._invalidate()
|
||||
return
|
||||
if event_type != "tool.started":
|
||||
return
|
||||
return self._show_tool_progress(event_type, preview, kwargs.get("percent"))
|
||||
if function_name and not function_name.startswith("_"):
|
||||
from agent.display import get_tool_preview_max_len, tool_row_emoji
|
||||
label = preview or function_name
|
||||
_pl = get_tool_preview_max_len()
|
||||
if _pl > 0 and len(label) > _pl:
|
||||
label = label[:_pl - 3] + "..."
|
||||
self._spinner_text = f"{tool_row_emoji(function_name, function_args)} {label}"
|
||||
self._spinner_text = self._tool_spinner_base = f"{tool_row_emoji(function_name, function_args)} {label}"
|
||||
self._tool_start_time = time.monotonic()
|
||||
# Store args for stacked scrollback line on completion
|
||||
self._pending_tool_info.setdefault(function_name, []).append(
|
||||
function_args if function_args is not None else {})
|
||||
self._invalidate()
|
||||
|
||||
def _show_tool_progress(self, event_type, message, percent) -> None:
|
||||
"""Live MCP progress rides the running tool's spinner line: ``<tool label> · 42% · message``."""
|
||||
base = self.__dict__.get("_tool_spinner_base")
|
||||
if event_type == "tool.progress" and base:
|
||||
parts = [base] + ([f"{percent}%"] if percent is not None else []) + ([message] if message else [])
|
||||
self._spinner_text = " · ".join(parts)
|
||||
self._invalidate()
|
||||
|
||||
def _on_tool_start(self, tool_call_id: str, function_name: str, function_args: dict):
|
||||
"""Capture local before-state for write-capable tools."""
|
||||
from cli import logger
|
||||
|
||||
@@ -952,6 +952,12 @@ class GatewayAgent:
|
||||
if self.tool_gen_callback is not None:
|
||||
self.tool_gen_callback(payload.get("name"))
|
||||
|
||||
def _on_tool_progress(self, payload, stream_callback, starts):
|
||||
if self.tool_progress_callback is not None:
|
||||
self.tool_progress_callback("tool.progress", payload.get("name"), payload.get("message"), None,
|
||||
tool_call_id=payload.get("tool_id"), progress=payload.get("progress"),
|
||||
total=payload.get("total"), percent=payload.get("percent"))
|
||||
|
||||
def _on_tool_start(self, payload, stream_callback, starts):
|
||||
self._close_segment(starts)
|
||||
name = payload.get("name") or payload.get("tool_name") or "tool"
|
||||
@@ -1386,6 +1392,7 @@ _HANDLERS = {
|
||||
"message.delta": GatewayAgent._on_delta,
|
||||
"message.interim": GatewayAgent._on_interim,
|
||||
"tool.generating": GatewayAgent._on_tool_generating,
|
||||
"tool.progress": GatewayAgent._on_tool_progress,
|
||||
"tool.start": GatewayAgent._on_tool_start,
|
||||
"tool.complete": GatewayAgent._on_tool_complete,
|
||||
"todo.updated": GatewayAgent._on_todo,
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
"""An MCP server's ``notifications/progress`` during ``tools/call`` reaches the agent's
|
||||
``tool_progress_callback`` as throttled ``tool.progress`` events for that call id. Real stdio child,
|
||||
real config, real SDK progress token round trip."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
SERVER = textwrap.dedent('''\
|
||||
import asyncio
|
||||
from mcp.server.mcpserver import MCPServer, Context
|
||||
app = MCPServer("prog")
|
||||
|
||||
@app.tool()
|
||||
async def slow(ctx: Context) -> str:
|
||||
for i in range(1, 21):
|
||||
await asyncio.sleep(0.1)
|
||||
await ctx.report_progress(i * 5, 100, f"stage {i}")
|
||||
return "done"
|
||||
|
||||
app.run()
|
||||
''')
|
||||
|
||||
|
||||
@pytest.mark.no_isolate
|
||||
@pytest.mark.platforms("posix")
|
||||
def test_mcp_progress_notifications_reach_tool_progress_callback(monkeypatch, tmp_path):
|
||||
pytest.importorskip("mcp.server.mcpserver")
|
||||
home = tmp_path / "home"
|
||||
home.mkdir()
|
||||
(home / "server.py").write_text(SERVER, encoding="utf-8")
|
||||
(home / "config.yaml").write_text(
|
||||
f"mcp_servers:\n prog:\n command: {json.dumps(sys.executable)}\n args: [{json.dumps(str(home / 'server.py'))}]\n",
|
||||
encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
|
||||
from agent.tool_progress_relay import bound_tool_progress
|
||||
from tools import mcp_tool_discovery as disc
|
||||
from tools import mcp_tool_loop as _loop
|
||||
from tools.registry import registry
|
||||
|
||||
events = []
|
||||
try:
|
||||
assert "mcp__prog__slow" in disc.discover_mcp_tools()
|
||||
with bound_tool_progress(lambda *a, **kw: events.append((a, kw)), "mcp__prog__slow", "call-7"):
|
||||
result = registry.dispatch("mcp__prog__slow", {})
|
||||
assert "done" in result
|
||||
finally:
|
||||
from tools.mcp_tool_lifecycle import shutdown_mcp_servers
|
||||
shutdown_mcp_servers()
|
||||
_loop._stop_mcp_loop()
|
||||
|
||||
assert events and all(a[0] == "tool.progress" and a[1] == "mcp__prog__slow" for a, _ in events)
|
||||
assert {kw["tool_call_id"] for _, kw in events} == {"call-7"}
|
||||
# 20 notifications over ~2 s, throttled to <= one per 500 ms, the final 100% always delivered.
|
||||
assert 2 <= len(events) <= 8
|
||||
assert events[-1][1]["percent"] == 100 and events[-1][0][2] == "stage 20"
|
||||
@@ -379,7 +379,7 @@ async def _track_inflight_rpc(server: Any, server_name: str, op: str, *, retry_s
|
||||
inflight.discard(task)
|
||||
|
||||
|
||||
async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str, args: dict):
|
||||
async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str, args: dict, progress_callback=None):
|
||||
"""``session.call_tool`` that fails fast when the stdio child is/gets dead: pre-call (a dead
|
||||
child must not hold the slot for the full timeout) and mid-call (race against
|
||||
``_watch_stdio_children``). Both raise :class:`_StdioChildExited` for the respawn path, which
|
||||
@@ -393,7 +393,10 @@ async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str
|
||||
f"MCP stdio subprocess for '{server_name}' had already exited when the call was dispatched",
|
||||
in_flight=False,
|
||||
)
|
||||
_call_coro = server.session.call_tool(tool_name, arguments=args)
|
||||
# A progress callback makes the SDK send ``_meta.progressToken``, the spec's opt-in for
|
||||
# ``notifications/progress``; without one the server must stay silent.
|
||||
extra = {"progress_callback": progress_callback} if progress_callback is not None else {}
|
||||
_call_coro = server.session.call_tool(tool_name, arguments=args, **extra)
|
||||
_watch_children = getattr(server, "_watch_stdio_children", None)
|
||||
if not (inspect.iscoroutinefunction(_watch_children) and asyncio.iscoroutine(_call_coro)):
|
||||
# Stubbed sessions return a non-awaitable, or there is no child-watcher to race: plain await.
|
||||
@@ -582,12 +585,20 @@ def _make_tool_handler(server_name: str, tool_name: str, tool_timeout: float, *,
|
||||
# pre-dispatch so the auth recoverer keeps its retry for every tool.
|
||||
read_only = _tool_is_read_only(server_name, tool_name)
|
||||
image_paths: list[str] = [] # the LAST attempt's cached images (a recoverer may retry _call)
|
||||
# Read on the agent's thread: the coroutine below runs on the MCP loop, outside this context.
|
||||
from agent.tool_progress_relay import current_tool_progress
|
||||
report = current_tool_progress()
|
||||
|
||||
async def _on_progress(progress, total, message):
|
||||
if report is not None:
|
||||
report(progress, total, message)
|
||||
|
||||
async def _call():
|
||||
async with server._rpc_lock, _track_inflight_rpc(server, server_name, op, retry_safe=read_only):
|
||||
server._pending_call_context = contextvars.copy_context() # for the elicitation callback
|
||||
try:
|
||||
result = await _call_tool_racing_stdio_death(server, server_name, tool_name, args)
|
||||
result = await _call_tool_racing_stdio_death(
|
||||
server, server_name, tool_name, args, _on_progress if report is not None else None)
|
||||
finally:
|
||||
server._pending_call_context = None
|
||||
if getattr(server, "_mark_session_proven", None) is not None: # round-trip done: transport healthy
|
||||
|
||||
@@ -409,6 +409,21 @@ class ToolOutputRiskPayload(Payload):
|
||||
event("tool.output_risk", ToolOutputRiskPayload, doc="Tool output was classified as risky (prompt-injection / secret findings).")
|
||||
|
||||
|
||||
class ToolProgressPayload(Payload):
|
||||
"""``agent.tool_progress_relay.tool_progress_payload`` — an MCP server's
|
||||
``notifications/progress`` for a running call, throttled to one per 500 ms per call."""
|
||||
|
||||
tool_id: str
|
||||
name: str
|
||||
progress: float
|
||||
total: float | None = None
|
||||
percent: int | None = None
|
||||
message: str | None = None
|
||||
|
||||
|
||||
event("tool.progress", ToolProgressPayload, doc="Live progress of a running tool call (MCP notifications/progress).")
|
||||
|
||||
|
||||
class TodoUpdatedPayload(Payload):
|
||||
"""``tool_progress._normalize_todo_state`` — full task snapshot."""
|
||||
|
||||
@@ -846,7 +861,8 @@ __all__ = [
|
||||
"SetupReadyPayload", "SideAgentCompletePayload", "SkinPayload", "StatusUpdatePayload",
|
||||
"StreamDeltaPayload", "SubagentEventPayload", "SubagentOutputTailEntry", "TerminalClosePayload",
|
||||
"TerminalOutputPayload", "TipShowPayload", "TodoUpdatedPayload", "ToolCompletePayload",
|
||||
"ToolGeneratingPayload", "ToolLabel", "ToolLabelKind", "ToolOutputRiskPayload", "ToolStartPayload",
|
||||
"ToolGeneratingPayload", "ToolLabel", "ToolLabelKind", "ToolOutputRiskPayload", "ToolProgressPayload",
|
||||
"ToolStartPayload",
|
||||
"TurnStatus", "VoicePartialPayload", "VoiceStatusPayload",
|
||||
"VoiceTranscriptPayload", "WakeDetectedPayload",
|
||||
]
|
||||
@@ -380,6 +380,13 @@ def _on_tool_complete(sid: str, tool_call_id: str, name: str, args: dict, result
|
||||
# `tool.started` is dropped on purpose: _on_tool_start already emits the authoritative tool.start with
|
||||
# the stable id and args; an id-less duplicate row makes the desktop live view diverge from history.
|
||||
|
||||
def _progress_tool_progress(sid, name, preview, kw):
|
||||
# Live MCP progress on a running tool row: tool chrome, so it follows display.tool_progress.
|
||||
if _tool_progress_enabled(sid):
|
||||
from agent.tool_progress_relay import tool_progress_payload
|
||||
_emit("tool.progress", sid, tool_progress_payload(name, preview, kw))
|
||||
|
||||
|
||||
def _progress_output_risk(sid, name, preview, kw):
|
||||
# A risk badge on a tool row: tool chrome, so it follows display.tool_progress.
|
||||
metadata = kw.get("risk_metadata")
|
||||
@@ -494,7 +501,8 @@ def _progress_subagent(sid: str, name: str, preview, kw, event_type):
|
||||
# event_type -> (handler, requires): `requires` names the arg that must be truthy for the row to be
|
||||
# emitted at all ("name" / "preview" / None).
|
||||
_PROGRESS_HANDLERS = {
|
||||
"tool.output_risk": (_progress_output_risk, "name"), "reasoning.available": (_progress_reasoning, "preview"),
|
||||
"tool.output_risk": (_progress_output_risk, "name"), "tool.progress": (_progress_tool_progress, "name"),
|
||||
"reasoning.available": (_progress_reasoning, "preview"),
|
||||
"moa.reference": (_progress_moa_reference, "name"),
|
||||
# Answer-only drops MoA content (references, progress/phase lines that land in the reasoning
|
||||
# disclosure or activity log) but keeps this: a bare state transition both clients use only
|
||||
|
||||
@@ -1223,6 +1223,35 @@ listener can't affect app dispatch. Every `host` door is async-safe: a sync thro
|
||||
from an internal helper (e.g. no desktop bridge in a plain browser) becomes a
|
||||
rejection your `.catch()` sees, never an error-boundary crash.
|
||||
|
||||
#### Live tool progress — `tool.progress` {#tool-progress-event}
|
||||
|
||||
While a tool call runs, an MCP server that sends `notifications/progress` (the
|
||||
MCP spec's progress channel; Hermes opts in by sending `_meta.progressToken` on
|
||||
every `tools/call`) reaches the renderer as a `tool.progress` event. The native
|
||||
tool row shows it as `42% · <message>` next to its timer; a plugin can render
|
||||
its own view:
|
||||
|
||||
```javascript
|
||||
ctx.onEvent('tool.progress', ({ session_id, payload }) => {
|
||||
// payload: { tool_id, name, progress, total, percent, message }
|
||||
if (payload.name.startsWith('mcp__meshy__')) showRefineProgress(payload.tool_id, payload.percent, payload.message)
|
||||
})
|
||||
ctx.onEvent('tool.complete', ({ payload }) => clearRefineProgress(payload.tool_id))
|
||||
```
|
||||
|
||||
| Field | Type | Meaning |
|
||||
|---|---|---|
|
||||
| `tool_id` | `string` | The call id; the same `tool_id` as the call's `tool.start` / `tool.complete` |
|
||||
| `name` | `string` | Registered tool name (`mcp__<server>__<tool>`) |
|
||||
| `progress` | `number` | The server's raw progress value |
|
||||
| `total` | `number \| null` | The server's total, when it sent one |
|
||||
| `percent` | `number \| null` | `0`–`100` (rounded, clamped) when `total` is set, else `null` |
|
||||
| `message` | `string \| null` | The server's progress message |
|
||||
|
||||
Events are throttled to at most one per 500 ms per call; the first update and
|
||||
the one that reaches `total` always arrive. They are never stored in the
|
||||
transcript, so a reopened session shows only the finished row.
|
||||
|
||||
`ctx.os` is the curated OS door — every way a plugin reaches outside the app
|
||||
window, in one namespace attributed to your plugin. `ctx.os.notify` posts a
|
||||
**native OS notification** — the same Electron pipeline the app's own
|
||||
|
||||
Reference in new issue
Block a user