Add detailed logging around docket.add() and subscription tasks

Tracing where the Redis ACL error occurs - the initial Redis writes
succeed but error happens somewhere after.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Guidry 2026-01-09 13:37:09 -05:00
commit c72b5dd7f0
2 changed files with 42 additions and 5 deletions

View file

@ -139,18 +139,38 @@ async def handle_tool_as_task(
# Queue function to Docket by name (result storage via execution_ttl)
# Use tool.key which matches what was registered - prefixed for mounted tools
await docket.add(
tool.key,
key=task_key,
)(**arguments)
_logger.info(
f"[{instance_id}] About to call docket.add: tool.key={tool.key}, "
f"task_key={task_key}, arguments={arguments}"
)
try:
await docket.add(
tool.key,
key=task_key,
)(**arguments)
_logger.info(f"[{instance_id}] docket.add completed successfully")
except Exception as e:
import traceback
_logger.error(
f"[{instance_id}] docket.add FAILED: {type(e).__name__}: {e}\n"
f"Traceback:\n{traceback.format_exc()}"
)
raise
# Spawn subscription task to send status notifications (SEP-1686 optional feature)
from fastmcp.server.tasks.subscriptions import subscribe_to_task_updates
# Start subscription in session's task group (persists for connection lifetime)
_logger.info(
f"[{instance_id}] Checking for subscription task group: "
f"hasattr={hasattr(ctx.session, '_subscription_task_group')}"
)
if hasattr(ctx.session, "_subscription_task_group"):
tg = ctx.session._subscription_task_group # type: ignore[attr-defined]
_logger.info(f"[{instance_id}] Task group: {tg}")
if tg:
_logger.info(f"[{instance_id}] Starting subscription task")
tg.start_soon( # type: ignore[union-attr]
subscribe_to_task_updates,
server_task_id,
@ -158,7 +178,9 @@ async def handle_tool_as_task(
ctx.session,
docket,
)
_logger.info(f"[{instance_id}] Subscription task started")
_logger.info(f"[{instance_id}] About to return task stub")
# Return task stub
# Tasks MUST begin in "working" status per SEP-1686 final spec (line 381)
return mcp.types.CallToolResult(

View file

@ -42,13 +42,23 @@ async def subscribe_to_task_updates(
session: MCP ServerSession for sending notifications
docket: Docket instance for subscribing to execution events
"""
logger.info(
f"subscribe_to_task_updates STARTING: task_id={task_id}, task_key={task_key}"
)
try:
logger.info(
f"subscribe_to_task_updates: About to call docket.get_execution({task_key})"
)
execution = await docket.get_execution(task_key)
logger.info(
f"subscribe_to_task_updates: docket.get_execution returned {execution}"
)
if execution is None:
logger.warning(f"No execution found for task {task_id}")
return
# Subscribe to state and progress events from Docket
logger.info("subscribe_to_task_updates: About to subscribe to execution events")
async for event in execution.subscribe():
if event["type"] == "state":
# Send notifications/tasks/status when state changes
@ -70,7 +80,12 @@ async def subscribe_to_task_updates(
)
except Exception as e:
logger.warning(f"Subscription task failed for {task_id}: {e}", exc_info=True)
import traceback
logger.error(
f"subscribe_to_task_updates FAILED for {task_id}: {type(e).__name__}: {e}\n"
f"Traceback:\n{traceback.format_exc()}"
)
async def _send_status_notification(