YggTorrent MCP Server

Instructions

Implementation Reference

• ygg_torrent/mcp_server.py:69-74 (handler)

  MCP tool handler for get_magnet_link. This is the primary execution function for the tool, registered via @mcp.tool() decorator.

  @mcp.tool()
  def get_magnet_link(torrent_id: int) -> str | None:
      """Get the magnet link from YggTorrent for a specific torrent by id."""
      logger.info(f"Getting magnet link for torrent: {torrent_id}")
      magnet_link: str | None = ygg_api.get_magnet_link(torrent_id)
      return magnet_link or "Magnet link not found"

• ygg_torrent/wrapper/api_client.py:162-178 (helper)

  Core helper function in YggTorrentApi that implements the magnet link retrieval by downloading the torrent file and generating the magnet URI.

  def get_magnet_link(self, torrent_id: int) -> str | None:
      """
      Get the magnet link for a specific torrent.
  
      Args:
          torrent_id: The ID of the torrent.
  
      Returns:
          The magnet link as a string or None.
      """
      try:
          file_bytes = self._download_torrent_file_bytes(torrent_id)
          if file_bytes and isinstance(file_bytes, bytes):
              return make_magnet_from_torrent_bytes(file_bytes)
      except Exception as e:
          print(f"Failed to generate magnet link: {e}")
      return None

• ygg_torrent/wrapper/torrent_editor.py:18-43 (helper)

  Utility function that generates the magnet link from raw torrent file bytes by computing info hash, name, size, and adding trackers.

  def make_magnet_from_torrent_bytes(file_bytes: bytes) -> str:
      metadata = decode(file_bytes)
      subj = metadata[b"info"]  # type: ignore
      hashcontents = encode(subj)  # type: ignore
      digest = sha1(hashcontents).digest()
      b32hash = b32encode(digest).decode()
  
      if b"files" in subj:
          total_length = sum(f[b"length"] for f in subj[b"files"])
      else:
          total_length = subj[b"length"]
  
      tracker = f"http://tracker.p2p-world.net:8080/{YGG_PASSKEY}/announce"
      tracker2 = f"http://connect.maxp2p.org:8080/{YGG_PASSKEY}/announce"
      return (
          "magnet:?xt=urn:btih:"
          + b32hash
          + "&dn="
          + parse.quote(subj[b"name"].decode())
          + "&tr="
          + tracker
          + "&tr="
          + tracker2
          + "&xl="
          + str(total_length)
      )