MCP Server for notify to weixin / telegram / bark / lark

Instructions

Implementation Reference

• mcp_notify/wework.py:157-165 (handler)

  The handler function for the 'wework_app_send_file' MCP tool. It defines the input schema using Pydantic Field for the file URL and target user, and executes by calling the shared wework_send_media helper with type 'file'.

  @mcp.tool(
      title="企业微信应用号-发送文件消息",
      description="通过企业微信应用号发送发送文件消息",
  )
  def wework_app_send_file(
      url: str = Field(description="文件URL"),
      touser: str = FIELD_TO_USER,
  ):
      return wework_send_media(touser, url, "file")

• mcp_notify/wework.py:167-200 (helper)

  Helper function that implements the core logic for sending media (image, video, voice, file) via WeWork app: downloads from URL, determines type, uploads to media API using access token, then sends the media_id to the specified user.

  def wework_send_media(touser, url: str, msgtype=None):
      if msgtype:
          pass
      elif '.jpg' in url.lower() or '.jpeg' in url.lower() or '.png' in url.lower():
          msgtype = 'image'
      elif '.mp4' in url.lower():
          msgtype = 'video'
      elif '.arm' in url.lower():
          msgtype = 'voice'
      else:
          msgtype = 'file'
      res = requests.get(url, timeout=120)
      res.raise_for_status()
      file = io.BytesIO(res.content)
      mine = res.headers.get("content-type") or "application/octet-stream"
      res = requests.post(
          f"{WEWORK_BASE_URL}/cgi-bin/media/upload",
          params={"type": msgtype, "access_token": get_access_token()},
          files={"media": ("filename", file, mine)},
          timeout=120,
      )
      media = res.json() or {}
      if not (media_id := media.get("media_id")):
          return media
      res = requests.post(
          f"{WEWORK_BASE_URL}/cgi-bin/message/send?access_token={get_access_token()}",
          json={
              "touser": touser or WEWORK_APP_TOUSER,
              "agentid": WEWORK_APP_AGENTID,
              "msgtype": msgtype,
              msgtype: {"media_id": media_id},
          },
      )
      return res.json()

• mcp_notify/wework.py:95-102 (helper)

  Cached helper function to obtain the WeWork application access token, required for API calls like media upload and message sending.

  @cached(TTLCache(maxsize=1, ttl=3600))
  def get_access_token():
      res = requests.get(
          f"{WEWORK_BASE_URL}/cgi-bin/gettoken",
          params={"corpid": WEWORK_APP_CORPID, "corpsecret": WEWORK_APP_SECRET},
          timeout=60,
      )
      return res.json().get("access_token")

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

  Calls add_tools from the wework module to register all WeWork-related tools, including 'wework_app_send_file', onto the FastMCP instance.

  wework.add_tools(mcp)