Notify MCP Server

Instructions

Implementation Reference

• mcp_notify/wework.py:151-155 (handler)

  The handler function for the 'wework_app_send_voice' tool. It takes a voice URL and optional touser, then delegates to the wework_send_media helper with msgtype='voice' to upload and send the voice message via the WeWork application API.

  def wework_app_send_voice(
      url: str = Field(description="语音URL"),
      touser: str = FIELD_TO_USER,
  ):
      return wework_send_media(touser, url, "voice")

• mcp_notify/wework.py:147-150 (registration)

  The @mcp.tool decorator that registers the 'wework_app_send_voice' tool with its title and description within the add_tools function.

  @mcp.tool(
      title="企业微信应用号-发送语音消息",
      description="通过企业微信应用号发送发送语音消息",
  )

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

  Core helper function that downloads media from URL, uploads it to WeWork media API using access token, then sends the message using the media_id. Determines msgtype based on URL extension or provided argument; used by voice, image, video, file tools.

  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 retrieve the WeWork application access token from corpid and secret, used in 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/wework.py:21-21 (schema)

  Pydantic Field definition for the 'touser' parameter shared across WeWork app tools.

  FIELD_TO_USER = Field("", description="接收消息的成员ID，多个用`|`分隔，为`@all`时向该企业应用全部成员发送，默认从环境变量获取")