aiogram-mcp

Instructions

Implementation Reference

• aiogram_mcp/tools/events.py:80-108 (handler)

  The handler implementation for the unsubscribe_events tool, which removes a subscription via the event manager.

  @mcp.tool
  async def unsubscribe_events(subscription_id: str) -> UnsubscribeEventsResult:
      """Unsubscribe from real-time Telegram events.
  
      Args:
          subscription_id: The subscription ID returned by subscribe_events.
      """
      audit_args = {"subscription_id": subscription_id}
  
      if ctx.event_manager is None:
          result = UnsubscribeEventsResult(
              ok=False,
              error="EventManager is not configured.",
          )
          if ctx.audit_logger:
              ctx.audit_logger.log("unsubscribe_events", audit_args, result.ok, result.error)
          return result
  
      removed = ctx.event_manager.unsubscribe(subscription_id)
      if removed:
          result = UnsubscribeEventsResult(ok=True, subscription_id=subscription_id)
      else:
          result = UnsubscribeEventsResult(
              ok=False,
              error=f"Subscription '{subscription_id}' not found.",
          )
      if ctx.audit_logger:
          ctx.audit_logger.log("unsubscribe_events", audit_args, result.ok, result.error)
      return result

• aiogram_mcp/tools/events.py:80-108 (handler)

  The handler for the 'unsubscribe_events' MCP tool.

  @mcp.tool
  async def unsubscribe_events(subscription_id: str) -> UnsubscribeEventsResult:
      """Unsubscribe from real-time Telegram events.
  
      Args:
          subscription_id: The subscription ID returned by subscribe_events.
      """
      audit_args = {"subscription_id": subscription_id}
  
      if ctx.event_manager is None:
          result = UnsubscribeEventsResult(
              ok=False,
              error="EventManager is not configured.",
          )
          if ctx.audit_logger:
              ctx.audit_logger.log("unsubscribe_events", audit_args, result.ok, result.error)
          return result
  
      removed = ctx.event_manager.unsubscribe(subscription_id)
      if removed:
          result = UnsubscribeEventsResult(ok=True, subscription_id=subscription_id)
      else:
          result = UnsubscribeEventsResult(
              ok=False,
              error=f"Subscription '{subscription_id}' not found.",
          )
      if ctx.audit_logger:
          ctx.audit_logger.log("unsubscribe_events", audit_args, result.ok, result.error)
      return result

• aiogram_mcp/tools/events.py:18-19 (schema)

  The result schema for the 'unsubscribe_events' tool.

  class UnsubscribeEventsResult(ToolResponse):
      subscription_id: str | None = None

• aiogram_mcp/tools/events.py:18-19 (schema)

  The result schema for the unsubscribe_events tool.

  class UnsubscribeEventsResult(ToolResponse):
      subscription_id: str | None = None

• aiogram_mcp/tools/events.py:78-78 (registration)

  Registration conditional check for 'unsubscribe_events'.

  if allowed_tools is None or "unsubscribe_events" in allowed_tools:

• aiogram_mcp/tools/events.py:78-78 (registration)

  The conditional registration logic for the unsubscribe_events tool within register_event_tools.

  if allowed_tools is None or "unsubscribe_events" in allowed_tools: