Telegram send audio

Instructions

Implementation Reference

• mcp_notify/tgbot.py:117-148 (registration)

  The tool 'tg_send_audio' is registered via the @mcp.tool decorator with title 'Telegram send audio' and description 'Send audio via telegram bot'. The decorator is how FastMCP registers tools.

  @mcp.tool(
      title="Telegram send audio",
      description="Send audio via telegram bot",
  )
  async def tg_send_audio(
      audio: str = Field(description="Audio URL or base64 data URI (e.g., data:audio/wav;base64,..."),
      chat_id: str = Field("", description="Telegram chat id, Default to get from environment variables"),
      caption: str = Field("", description="Audio caption, 0-1024 characters after entities parsing"),
      parse_mode: str = Field("", description=f"Mode for parsing entities in the caption. [text/MarkdownV2]"),
      reply_to_message_id: int = Field(0, description="Identifier of the message that will be replied to"),
  ):
      if parse_mode == TELEGRAM_MARKDOWN_V2:
          caption = telegramify_markdown.markdownify(caption)
  
      if audio.startswith("data:"):
          match = re.match(r"data:audio/([^;]+);base64,(.*)", audio)
          if not match:
              return {"error": "Invalid base64 data URL format"}
          try:
              datas = base64.b64decode(match.group(2))
              audio = InputFile(io.BytesIO(datas), f"audio.{match.group(1)}")
          except Exception as e:
              return {"error": f"Failed to decode base64: {str(e)}"}
  
      res = await bot.send_audio(
          chat_id=chat_id or TELEGRAM_DEFAULT_CHAT,
          audio=audio,
          caption=caption or None,
          parse_mode=parse_mode if parse_mode in [TELEGRAM_MARKDOWN_V2] else None,
          reply_to_message_id=reply_to_message_id or None,
      )
      return res.to_json()

• mcp_notify/tgbot.py:121-148 (handler)

  The tg_send_audio async function is the handler. It accepts audio (URL or base64 data URI), chat_id, caption, parse_mode, and reply_to_message_id. It decodes base64 data URIs for audio files and calls bot.send_audio() to send the audio via Telegram.

  async def tg_send_audio(
      audio: str = Field(description="Audio URL or base64 data URI (e.g., data:audio/wav;base64,..."),
      chat_id: str = Field("", description="Telegram chat id, Default to get from environment variables"),
      caption: str = Field("", description="Audio caption, 0-1024 characters after entities parsing"),
      parse_mode: str = Field("", description=f"Mode for parsing entities in the caption. [text/MarkdownV2]"),
      reply_to_message_id: int = Field(0, description="Identifier of the message that will be replied to"),
  ):
      if parse_mode == TELEGRAM_MARKDOWN_V2:
          caption = telegramify_markdown.markdownify(caption)
  
      if audio.startswith("data:"):
          match = re.match(r"data:audio/([^;]+);base64,(.*)", audio)
          if not match:
              return {"error": "Invalid base64 data URL format"}
          try:
              datas = base64.b64decode(match.group(2))
              audio = InputFile(io.BytesIO(datas), f"audio.{match.group(1)}")
          except Exception as e:
              return {"error": f"Failed to decode base64: {str(e)}"}
  
      res = await bot.send_audio(
          chat_id=chat_id or TELEGRAM_DEFAULT_CHAT,
          audio=audio,
          caption=caption or None,
          parse_mode=parse_mode if parse_mode in [TELEGRAM_MARKDOWN_V2] else None,
          reply_to_message_id=reply_to_message_id or None,
      )
      return res.to_json()

• mcp_notify/tgbot.py:121-127 (schema)

  The input parameters are defined via Pydantic Field descriptors: audio (str), chat_id (str), caption (str), parse_mode (str), reply_to_message_id (int). These serve as the input schema for the tool.

  async def tg_send_audio(
      audio: str = Field(description="Audio URL or base64 data URI (e.g., data:audio/wav;base64,..."),
      chat_id: str = Field("", description="Telegram chat id, Default to get from environment variables"),
      caption: str = Field("", description="Audio caption, 0-1024 characters after entities parsing"),
      parse_mode: str = Field("", description=f"Mode for parsing entities in the caption. [text/MarkdownV2]"),
      reply_to_message_id: int = Field(0, description="Identifier of the message that will be replied to"),
  ):

• mcp_notify/tgbot.py:16-21 (registration)

  The add_tools function creates the Bot instance with TELEGRAM_BOT_TOKEN and configures the base URLs. This is the setup/registration entry point called from __init__.py (line 20: tgbot.add_tools(mcp)).

  def add_tools(mcp: FastMCP, logger=None):
      bot = Bot(
          TELEGRAM_BOT_TOKEN,
          base_url=f"{TELEGRAM_BASE_URL}/bot",
          base_file_url=f"{TELEGRAM_BASE_URL}/file/bot",
      ) if TELEGRAM_BOT_TOKEN else None

• mcp_notify/__init__.py:20-20 (registration)

  The registration call chain: __init__.py calls tgbot.add_tools(mcp) which registers all Telegram tools including tg_send_audio.

  tgbot.add_tools(mcp)