subscribe_events

Subscribe to real-time Telegram events with filters for chats and event types. Receive notifications when new events match your criteria and read event data from the queue resource.

Instructions

Subscribe to real-time Telegram events.

Receive notifications when new events match your filters. Read the telegram://events/queue resource to get event data.

Args: chat_ids: Only events from these chats (None = all allowed chats). event_types: Filter by type: "message", "command" (None = all).

Input Schema

TableJSON Schema

Name	Required	Description	Default
`chat_ids`	No
`event_types`	No

Implementation Reference

aiogram_mcp/tools/events.py:27-76 (handler)

The handler function for the `subscribe_events` tool. It validates the request, checks chat permissions via context, and uses the `event_manager` to perform the subscription.

@mcp.tool
async def subscribe_events(
    chat_ids: list[int] | None = None,
    event_types: list[str] | None = None,
) -> SubscribeEventsResult:
    """Subscribe to real-time Telegram events.

    Receive notifications when new events match your filters.
    Read the telegram://events/queue resource to get event data.

    Args:
        chat_ids: Only events from these chats (None = all allowed chats).
        event_types: Filter by type: "message", "command" (None = all).
    """
    audit_args = {"chat_ids": chat_ids, "event_types": event_types}

    if ctx.event_manager is None:
        result = SubscribeEventsResult(
            ok=False,
            error="EventManager is not configured. Pass event_manager to AiogramMCP.",
        )
        if ctx.audit_logger:
            ctx.audit_logger.log("subscribe_events", audit_args, result.ok, result.error)
        return result

    if chat_ids is not None:
        for cid in chat_ids:
            if not ctx.is_chat_allowed(cid):
                result = SubscribeEventsResult(
                    ok=False,
                    error=f"Chat {cid} is not in allowed_chat_ids.",
                )
                if ctx.audit_logger:
                    ctx.audit_logger.log("subscribe_events", audit_args, result.ok, result.error)
                return result

    sub_id = ctx.event_manager.subscribe(
        chat_ids=chat_ids,
        event_types=event_types,
    )
    result = SubscribeEventsResult(
        ok=True,
        subscription_id=sub_id,
        chat_ids=chat_ids,
        event_types=event_types,
        note="Read telegram://events/queue to get events.",
    )
    if ctx.audit_logger:
        ctx.audit_logger.log("subscribe_events", audit_args, result.ok, result.error)
    return result

aiogram_mcp/tools/events.py:11-15 (schema)

The schema defining the response structure for `subscribe_events`.

class SubscribeEventsResult(ToolResponse):
    subscription_id: str | None = None
    chat_ids: list[int] | None = None
    event_types: list[str] | None = None
    note: str | None = None

aiogram_mcp/tools/events.py:25-25 (registration)
Registration logic for the `subscribe_events` tool within the `register_event_tools` function.
```
if allowed_tools is None or "subscribe_events" in allowed_tools:
```

aiogram-mcp