pin_message

Instructions

Implementation Reference

• aiogram_mcp/tools/messaging.py:225-261 (handler)

  The pin_message tool handler function. Registers an MCP tool that pins a message in a chat via ctx.bot.pin_chat_message(). Checks chat permissions, applies rate limiting, and logs to audit.

  async def pin_message(
      chat_id: int,
      message_id: int,
      disable_notification: bool = False,
  ) -> OkResult:
      """Pin a message in a chat."""
      if not ctx.is_chat_allowed(chat_id):
          result = OkResult(ok=False, error=f"Chat {chat_id} is not allowed.")
          if ctx.audit_logger:
              ctx.audit_logger.log(
                  "pin_message",
                  {"chat_id": chat_id, "message_id": message_id},
                  result.ok,
                  result.error,
              )
          return result
  
      try:
          if ctx.rate_limiter:
              await ctx.rate_limiter.acquire()
          await ctx.bot.pin_chat_message(
              chat_id=chat_id,
              message_id=message_id,
              disable_notification=disable_notification,
          )
          result = OkResult(ok=True)
      except (TelegramBadRequest, TelegramForbiddenError) as exc:
          result = OkResult(ok=False, error=str(exc))
  
      if ctx.audit_logger:
          ctx.audit_logger.log(
              "pin_message",
              {"chat_id": chat_id, "message_id": message_id},
              result.ok,
              result.error,
          )
      return result

• aiogram_mcp/tools/messaging.py:222-261 (registration)

  Conditional registration of pin_message tool within register_messaging_tools(). Checks if 'pin_message' is in the allowed_tools set before decorating with @mcp.tool.

  if allowed_tools is None or "pin_message" in allowed_tools:
  
      @mcp.tool
      async def pin_message(
          chat_id: int,
          message_id: int,
          disable_notification: bool = False,
      ) -> OkResult:
          """Pin a message in a chat."""
          if not ctx.is_chat_allowed(chat_id):
              result = OkResult(ok=False, error=f"Chat {chat_id} is not allowed.")
              if ctx.audit_logger:
                  ctx.audit_logger.log(
                      "pin_message",
                      {"chat_id": chat_id, "message_id": message_id},
                      result.ok,
                      result.error,
                  )
              return result
  
          try:
              if ctx.rate_limiter:
                  await ctx.rate_limiter.acquire()
              await ctx.bot.pin_chat_message(
                  chat_id=chat_id,
                  message_id=message_id,
                  disable_notification=disable_notification,
              )
              result = OkResult(ok=True)
          except (TelegramBadRequest, TelegramForbiddenError) as exc:
              result = OkResult(ok=False, error=str(exc))
  
          if ctx.audit_logger:
              ctx.audit_logger.log(
                  "pin_message",
                  {"chat_id": chat_id, "message_id": message_id},
                  result.ok,
                  result.error,
              )
          return result

• aiogram_mcp/models.py:15-18 (schema)

  OkResult model – the return type for pin_message. Extends ToolResponse with just ok/error fields.

  class OkResult(ToolResponse):
      """Result for tools that return only ok/error (delete, pin, etc.)."""
  
      pass

• aiogram_mcp/context.py:29-34 (helper)

  is_chat_allowed helper used by pin_message to verify the chat_id is in the whitelist.

  def is_chat_allowed(self, chat_id: int) -> bool:
      """Return whether the MCP server may act on a chat."""
      if self.allowed_chat_ids is None:
          return True
      return chat_id in self.allowed_chat_ids

• aiogram_mcp/server.py:86-88 (registration)

  Top-level registration in AiogramMCP._register_tools() which calls register_messaging_tools() with the allowed_tools set derived from permission level.

  def _register_tools(self) -> None:
      at = self._allowed_tools
      register_messaging_tools(self._mcp, self._ctx, allowed_tools=at)