M-Team MCP Server

Instructions

Implementation Reference

• mcp_server_mteam/server.py:205-277 (handler)

  The 'download_torrent' function is decorated with @mcp.tool and handles the logic of requesting a download token, fetching the torrent details, and saving the .torrent file to the local directory.

  @mcp.tool
  def download_torrent(torrent_id: str) -> str:
      """Download a .torrent file for the given torrent ID.
  
      This tool first obtains a temporary download token from the M-Team API,
      then downloads the .torrent file and saves it to the configured download
      directory (default: ./seed/).
  
      Args:
          torrent_id: The numeric torrent ID (e.g. "1125330").
  
      Returns:
          The absolute path to the saved .torrent file on success, or an error
          message on failure.
      """
      token_url = f"{BASE_URL}/torrent/genDlToken"
      token_headers = {
          "accept": "*/*",
          "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
          "x-api-key": API_KEY,
      }
      if not API_KEY:
          raise RuntimeError(
              "MTEAM_API_KEY is not set. "
              "Please add it to your .env file (see .env_example)."
          )
  
      resp = requests.post(
          token_url, headers=token_headers, data={"id": torrent_id}, timeout=15
      )
      resp.raise_for_status()
      token_data = resp.json()
  
      if token_data.get("message") != "SUCCESS" or not token_data.get("data"):
          return f"Failed to get download token: {token_data}"
  
      download_url: str = token_data["data"]
  
      # Fetch torrent name for a meaningful filename; fall back to ID if unavailable
      torrent_name = torrent_id
      try:
          detail_headers = {k: v for k, v in _headers().items() if k != "Content-Type"}
          detail_resp = requests.post(
              f"{BASE_URL}/torrent/detail",
              data={"id": torrent_id},
              headers=detail_headers,
              timeout=15,
          )
          detail_resp.raise_for_status()
          detail_data = detail_resp.json()
          raw_name = (detail_data.get("data") or {}).get("name") or ""
          if raw_name:
              torrent_name = raw_name
      except Exception:
          pass
  
      # Replace spaces with dots, then strip illegal filesystem characters
      dotted = torrent_name.replace(" ", ".")
      safe_name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", dotted).strip(".")
      if not safe_name:
          safe_name = torrent_id
      filename = f"[M-TEAM]{safe_name}.torrent"
      os.makedirs(DOWNLOAD_DIR, exist_ok=True)
      output_path = os.path.abspath(os.path.join(DOWNLOAD_DIR, filename))
  
      dl_resp = requests.get(download_url, stream=True, timeout=60)
      dl_resp.raise_for_status()
      with open(output_path, "wb") as f:
          for chunk in dl_resp.iter_content(chunk_size=128 * 1024):
              if chunk:
                  f.write(chunk)
  
      return f"Torrent saved to: {output_path}"