Notify MCP Server

Instructions

Implementation Reference

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

  Full handler function for the 'wework_app_send_file' MCP tool, including decorator, schema (parameters), and logic that delegates to shared media helper.

  @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)

  Shared helper function implementing the core logic for sending media files (image, video, voice, file) via enterprise WeChat app API, used by wework_app_send_file.

  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/__init__.py:19-19 (registration)

  Invocation of add_tools from wework module, which registers the wework_app_send_file tool (and others) to the FastMCP instance.

  wework.add_tools(mcp)

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

  Cached helper for obtaining WeWork app access token, used in media upload and send operations.

  @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")