py-libp2p

https://py-libp2p.readthedocs.io/en/stable/
Admin
py-libp2p is the Python implementation of the libp2p networking stack, providing a modular...
Tokens:752,289
Snippets:6,449
Trust Score:9.6
Update:1 day ago
Show doc for...
Tokens:
Raw
### Install dependencies

Source: https://py-libp2p.readthedocs.io/en/stable/examples.announce_addrs.html

Install the necessary dependencies from the root of the repository before running the example.

```bash
$ python -m pip install -e .
```

--------------------------------

### Install libp2p and run Random Walk server mode

Source: https://py-libp2p.readthedocs.io/en/stable/examples.random_walk.html

Install the libp2p library, then run the Random Walk example in server mode. The output shows the server starting on a port, the node peer ID and address, initial routing table size, and DHT service startup. After a demo interval, it displays routing table size, connected peers, and peerstore size.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ cd examples/random_walk
$ python random_walk.py --mode server
2025-08-12 19:51:25,424 - random-walk-example - INFO - === Random Walk Example for py-libp2p ===
2025-08-12 19:51:25,424 - random-walk-example - INFO - Mode: server, Port: 0 Demo interval: 30s
2025-08-12 19:51:25,426 - random-walk-example - INFO - Starting server node on port 45123
2025-08-12 19:51:25,426 - random-walk-example - INFO - Node peer ID: 16Uiu2HAm7EsNv5vvjPAehGAVfChjYjD63ZHyWogQRdzntSbAg9ef
2025-08-12 19:51:25,426 - random-walk-example - INFO - Node address: /ip4/127.0.0.1/tcp/45123/p2p/16Uiu2HAm7EsNv5vvjPAehGAVfChjYjD63ZHyWogQRdzntSbAg9ef
2025-08-12 19:51:25,427 - random-walk-example - INFO - Initial routing table size: 0
2025-08-12 19:51:25,427 - random-walk-example - INFO - DHT service started in SERVER mode
2025-08-12 19:51:25,430 - libp2p.discovery.random_walk.rt_refresh_manager - INFO - RT Refresh Manager started
2025-08-12 19:51:55,432 - random-walk-example - INFO - --- Iteration 1 ---
2025-08-12 19:51:55,432 - random-walk-example - INFO - Routing table size: 15
2025-08-12 19:51:55,432 - random-walk-example - INFO - Connected peers: 8
2025-08-12 19:51:55,432 - random-walk-example - INFO - Peerstore size: 42
```

--------------------------------

### Install libp2p and run chat-demo

Source: https://py-libp2p.readthedocs.io/en/stable/examples.chat.html

Install the libp2p package and start the chat demo. The output shows the installation and the command to run in another console.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ chat-demo
Run this from the same folder in another console:

chat-demo -p 8001 -d /ip4/127.0.0.1/tcp/8000/p2p/QmPouApKqyxJDy6YT21EXNS6efuNzvJ3W3kqRQxkQ77GFJ

Waiting for incoming connection...
```

--------------------------------

### Rendezvous Protocol Example (Python)

Source: https://py-libp2p.readthedocs.io/en/stable/examples.rendezvous.html

Complete example script showing how to start a rendezvous server and run a client that registers and discovers peers. Includes setup for logging, host creation, and error handling.

```python
#!/usr/bin/env python3
"""
Simple example demonstrating rendezvous protocol usage.

This example shows how to:
1. Start a rendezvous service
2. Register a peer under a namespace
3. Discover other peers in the same namespace
"""

import argparse
import logging
from pathlib import Path
import sys
import traceback

# Add parent directory to path to import libp2p
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

import multiaddr
import trio

from libp2p import new_host
from libp2p.discovery.rendezvous import (
    RendezvousDiscovery,
    RendezvousService,
    config,
)
from libp2p.peer.peerinfo import PeerInfo, info_from_p2p_addr

# Enable logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()],
)
logging.getLogger().setLevel(logging.INFO)
logging.getLogger("libp2p.discovery.rendezvous").setLevel(logging.INFO)
logging.getLogger("libp2p").propagate = True

# Create logger for this example
logger = logging.getLogger("rendezvous_example")


async def run_rendezvous_server(port: int = 0):
    """Run a rendezvous server."""
    listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
    host = new_host()

    async with host.run([listen_addr]), trio.open_nursery() as nursery:
        # Start the peer-store cleanup task
        nursery.start_soon(host.get_peerstore().start_cleanup_task, 60)

        # Start rendezvous service
        service = RendezvousService(host)

        actual_addrs = host.get_addrs()
        logger.info(f"Rendezvous server started with peer ID: {host.get_id()}")
        logger.info(
            f"Listening on: {actual_addrs[0] if actual_addrs else 'no addresses'}"
        )
        logger.info("To connect a client, use:")
        if actual_addrs:
            logger.info(
                f"  python rendezvous.py --mode client --address {actual_addrs[0]}"
            )
        logger.info("Press Ctrl+C to stop...")

        try:
            # Keep server running and print stats periodically
            while True:
                await trio.sleep(10)
                stats = service.get_namespace_stats()
                if stats:
                    logger.info(f"Namespace stats: {stats}")
                else:
                    logger.info("No active registrations")
        except KeyboardInterrupt:
            logger.info("Shutting down rendezvous server...")
        except Exception as e:
            logger.error(f"Unexpected error in server: {e}")
            raise


async def run_client_example(
    server_addr: str,
    namespace: str = config.DEFAULT_NAMESPACE,
    enable_refresh: bool = False,
    port: int = 0,
):
    """Run a client that registers and discovers peers."""
    listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
    host = new_host()

    # Parse server address and extract peer info
    try:
        server_maddr = multiaddr.Multiaddr(server_addr)
        server_info = info_from_p2p_addr(server_maddr)
    except Exception as e:
        logger.error(f"Failed to parse server address '{server_addr}': {e}")
        return

    async with host.run([listen_addr]), trio.open_nursery() as nursery:
        # Start the peer-store cleanup task
        nursery.start_soon(host.get_peerstore().start_cleanup_task, 60)

        # Connect to server
        try:
            await host.connect(server_info)
            logger.info(f"Connected to rendezvous server: {server_info.peer_id}")
            logger.info(f"Enable refresh: {enable_refresh}")
        except Exception as e:
            logger.error(f"Failed to connect to server: {e}")
            return

        # Create rendezvous discovery
        discovery = RendezvousDiscovery(host, server_info.peer_id, enable_refresh)

        # Run discovery service in background if refresh is enabled
        async with trio.open_nursery() as nursery:
            if enable_refresh:
                # Start the discovery service
                nursery.start_soon(discovery.run)
                logger.info(
                    "🔄 Refresh mode enabled - discovery service running in background"
                )

            try:
                logger.info(f"Client started with peer ID: {host.get_id()}")

                # Register under a namespace with optional auto-refresh
                logger.info(f"Registering in namespace '{namespace}'...")
```

--------------------------------

### Install libp2p and run identify demo (first host)

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify.html

Install the libp2p package and start the first host. The output shows the command to run from another console to connect a second host.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ identify-demo
First host listening. Run this from another console:

identify-demo -p 8889 -d /ip4/127.0.0.1/tcp/8888/p2p/QmUiN4R3fNrCoQugGgmmb3v35neMEjKFNrsbNGVDsRHWpM

Waiting for incoming identify request...

```

--------------------------------

### examples.tls.example_tls_server.main()

Source: https://py-libp2p.readthedocs.io/en/stable/examples.tls.html

Entry point for the TLS chat server example. Parses command-line arguments and starts the server.

```APIDOC
## examples.tls.example_tls_server.main()

### Description
Entry point for the TLS chat server example. Parses command-line arguments and starts the server.

### Method
Function

### Signature
`main() -> None`

### Parameters
None

### Returns
None

```

--------------------------------

### Main setup for identify-push example

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify_push.html

Sets up two libp2p hosts with key pairs and registers custom identify and identify/push handlers. This is the entry point for the demo.

```python
async def main() -> None:
    print("\n==== Starting Enhanced Identify-Push Example ====\n")

    # Create key pairs for the two hosts
    key_pair_1 = create_new_key_pair()
    key_pair_2 = create_new_key_pair()

    # Create the first host
    host_1 = new_host(key_pair=key_pair_1)

    # Set up custom identify and identify/push handlers
    host_1.set_stream_handler(
        TProtocol("/ipfs/id/1.0.0"), create_custom_identify_handler(host_1, "Host 1")
    )
    host_1.set_stream_handler(
        ID_PUSH, create_custom_identify_push_handler(host_1, "Host 1")
    )

    # Create the second host
    host_2 = new_host(key_pair=key_pair_2)
```

--------------------------------

### Linux setup with uv (Option 2)

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Set up the development environment using uv, which is the same as CI. Includes installing pre-commit hooks.

```bash
cd py-libp2p
uv sync --group dev
uv run pre-commit install
```

--------------------------------

### Install libp2p and start relay node

Source: https://py-libp2p.readthedocs.io/en/stable/examples.circuit_relay.html

Install the libp2p package and start a relay node with the circuit-relay-demo command. The relay node listens on port 8000 and prints its peer ID after 'Listening on:'. Copy that peer ID for use in subsequent commands.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ circuit-relay-demo --role relay --port 8000 --seed 1
Starting relay node...
Listening on: /ip4/0.0.0.0/tcp/8000/p2p/16Uiu2HAm765pEhEgJJmz5xHLt8Ay2a16qtS9uPzjUGjuCKTAtQYN
```

--------------------------------

### Linux setup with setup script (Option 1)

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Recommended setup script for Linux. If not already in a virtual environment, the script creates one and instructs you to activate it.

```bash
cd py-libp2p
./scripts/setup_dev.sh
```

--------------------------------

### Register example console scripts in setup.py

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Add the example as a console script entry point in setup.py. This makes the example runnable via a command like 'identify-demo'.

```python
entry_points={
    "console_scripts": [
        "chat-demo=examples.chat.chat:main",
        "echo-demo=examples.echo.echo:main",
        "ping-demo=examples.ping.ping:main",
        "identify-demo=examples.identify.identify:main",
        "circuit-relay-demo=examples.circuit_relay.relay_example:main"
    ],
}
```

--------------------------------

### Install libp2p and run Kademlia server

Source: https://py-libp2p.readthedocs.io/en/stable/examples.kademlia.html

Install the libp2p package and start the Kademlia server. The server stores a value and advertises content, printing a bootstrap address to connect clients.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ cd examples/kademlia
$ python kademlia.py --mode server
2025-06-13 19:51:25,424 - kademlia-example - INFO - Running in server mode on port 0
2025-06-13 19:51:25,426 - kademlia-example - INFO - Connected to bootstrap nodes: []
2025-06-13 19:51:25,426 - kademlia-example - INFO - To connect to this node, use: --bootstrap /ip4/127.0.0.1/tcp/28910/p2p/16Uiu2HAm7EsNv5vvjPAehGAVfChjYjD63ZHyWogQRdzntSbAg9ef
2025-06-13 19:51:25,426 - kademlia-example - INFO - Saved server address to log: /ip4/127.0.0.1/tcp/28910/p2p/16Uiu2HAm7EsNv5vvjPAehGAVfChjYjD63ZHyWogQRdzntSbAg9ef
2025-06-13 19:51:25,427 - kademlia-example - INFO - DHT service started in SERVER mode
2025-06-13 19:51:25,427 - kademlia-example - INFO - Stored value 'Hello message from Sumanjeet' with key: FVDjasarSFDoLPMdgnp1dHSbW2ZAfN8NU2zNbCQeczgP
2025-06-13 19:51:25,427 - kademlia-example - INFO - Successfully advertised as server for content: 361f2ed1183bca491b8aec11f0b9e5c06724759b0f7480ae7fb4894901993bc8
```

--------------------------------

### Start intro peer A

Source: https://py-libp2p.readthedocs.io/en/stable/examples.dht_chat.html

Run the DHT chat example as the intro peer on port 18001. The peer starts with an empty DHT and prints its Peer ID and listening multiaddr for others to bootstrap.

```bash
$ python -m examples.dht_chat.dht_chat --port 18001
```

--------------------------------

### Install libp2p and run echo-quic-demo

Source: https://py-libp2p.readthedocs.io/en/stable/examples.echo_quic.html

Install the libp2p package and run the echo-quic-demo command. The output shows successful installation and the server waiting for incoming connections.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ echo-quic-demo
Run this from the same folder in another console:

echo-quic-demo -d /ip4/127.0.0.1/udp/8000/quic-v1/p2p/16Uiu2HAmAsbxRR1HiGJRNVPQLNMeNsBCsXT3rDjoYBQzgzNpM5mJ

Waiting for incoming connection...

```

--------------------------------

### Install libp2p and run pubsub-demo (first node)

Source: https://py-libp2p.readthedocs.io/en/stable/examples.pubsub.html

Install the libp2p package with pip, then run the pubsub-demo command. The output shows the node starting, subscribing to the 'pubsub-chat' topic, and printing a command to connect from another terminal.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ pubsub-demo
2025-04-06 23:59:17,471 - pubsub-demo - INFO - Running pubsub chat example...
2025-04-06 23:59:17,471 - pubsub-demo - INFO - Your selected topic is: pubsub-chat
2025-04-06 23:59:17,472 - pubsub-demo - INFO - Using random available port: 33269
2025-04-06 23:59:17,490 - pubsub-demo - INFO - Node started with peer ID: QmcJnocH1d1tz3Zp4MotVDjNfNFawXHw2dpB9tMYGTXJp7
2025-04-06 23:59:17,490 - pubsub-demo - INFO - Listening on: /ip4/127.0.0.1/tcp/33269
2025-04-06 23:59:17,490 - pubsub-demo - INFO - Initializing PubSub and GossipSub...
2025-04-06 23:59:17,491 - pubsub-demo - INFO - Pubsub and GossipSub services started.
2025-04-06 23:59:17,491 - pubsub-demo - INFO - Pubsub ready.
2025-04-06 23:59:17,491 - pubsub-demo - INFO - Subscribed to topic: pubsub-chat
2025-04-06 23:59:17,491 - pubsub-demo - INFO - Run this script in another console with:
pubsub-demo -d /ip4/127.0.0.1/tcp/33269/p2p/QmcJnocH1d1tz3Zp4MotVDjNfNFawXHw2dpB9tMYGTXJp7

2025-04-06 23:59:17,491 - pubsub-demo - INFO - Waiting for peers...
Type messages to send (press Enter to send):

```

--------------------------------

### Install libp2p and run metrics-demo

Source: https://py-libp2p.readthedocs.io/en/stable/examples.metrics.html

Install the libp2p package and start the metrics demo node. The output shows the host multiaddr and confirms that Gossipsub, Pubsub, and DHT services are started. Prometheus metrics are exposed at http://localhost:8000.

```console
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x

$ metrics-demo
Host multiaddr: /ip4/172.16.68.73/tcp/41173/p2p/12D3KooWD2DFvDs4wekLWU8sAUJJgivbRbiiKkX9yQ3kGhuCwCqL
Gossipsub and Pubsub services started !!
DHT service started with DHTMode.SERVER mode
Starting command executor loop...

Prometheus metrics visible at: http://localhost:8000

To start prometheus and grafana dashboards, from another terminal:
PROMETHEUS_PORT=9001 GRAFANA_PORT=7001 docker compose up

After this:
Prometheus dashboard will be visible at: http://localhost:9001
Grafana dashboard will be visible at: http://localhost:7001

Entering intractive mode, type commands below.

Available commands:
- connect <multiaddr>               - Connect to another peer
...
```

--------------------------------

### Install libp2p and run ping-demo

Source: https://py-libp2p.readthedocs.io/en/stable/examples.ping.html

Install the libp2p package and run the ping-demo command. The output shows the installation and the command to run from another console.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ ping-demo
Run this from the same folder in another console:

ping-demo -p 8001 -d /ip4/127.0.0.1/tcp/8000/p2p/QmXfptdHU6hqG95JswxYVUH4bphcK8y18mhFcgUQFe6fCN

Waiting for incoming connection...

```

--------------------------------

### macOS setup with uv (Option 2)

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Set up the development environment using uv with CFLAGS and LDFLAGS for gmp. Includes installing pre-commit hooks.

```bash
cd py-libp2p
CFLAGS="$(pkg-config --cflags gmp)" LDFLAGS="$(pkg-config --libs gmp)" uv sync --group dev
uv run pre-commit install
```

--------------------------------

### Start listener node with announce addresses

Source: https://py-libp2p.readthedocs.io/en/stable/examples.announce_addrs.html

Run the example script as Node A (listener) with announce addresses to advertise publicly reachable addresses instead of the local listen address.

```bash
$ python examples/announce_addrs/announce_addrs.py --listen-port 9001 \
    --announce /dns4/example.ngrok-free.app/tcp/9001 /ip4/1.2.3.4/tcp/4001
```

--------------------------------

### Install libp2p and run echo-demo

Source: https://py-libp2p.readthedocs.io/en/stable/examples.echo.html

Install the libp2p package and run the echo-demo command. The output shows the installation and the command to run from another console.

```bash
$ python -m pip install libp2p
Collecting libp2p
...
Successfully installed libp2p-x.x.x
$ echo-demo
Run this from the same folder in another console:

echo-demo -p 8001 -d /ip4/127.0.0.1/tcp/8000/p2p/16Uiu2HAmAsbxRR1HiGJRNVPQLNMeNsBCsXT3rDjoYBQzgzNpM5mJ

Waiting for incoming connection...

```

--------------------------------

### start()

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.transport.webtransport.html

Starts the multiplexer.

```APIDOC
## start()

### Description
Start the multiplexer.

### Method
Async Function

### Signature
`async start() -> None`

### Returns
- **None**
```

--------------------------------

### Install and verify Git

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Install Git using winget and verify the installation.

```bash
winget install --id Git.Git -e
git --version
```

--------------------------------

### Basic GossipSub 1.2 Setup

Source: https://py-libp2p.readthedocs.io/en/stable/gossipsub-1.2.html

Initializes a libp2p node, creates a GossipSub instance with default settings, wraps it in a PubSub instance, and starts the node.

```python
from libp2p import new_node
from libp2p.pubsub.gossipsub import GossipSub
from libp2p.pubsub.pubsub import PubSub

# Create a new libp2p node
node = await new_node()

# Initialize GossipSub with default settings
gossipsub = GossipSub()

# Initialize PubSub with GossipSub
pubsub = PubSub(gossipsub)

# Start the node
await node.start()
```

--------------------------------

### Listener host setup and custom identify handler

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify.html

Sets up the first host as a listener, starts the peerstore cleanup task, prints listening addresses, and replaces the identify handler with a custom one that logs peer ID and remote address. The custom handler calls the original handler after logging.

```python
host_a, use_varint_format=use_varint_format
    )
    host_a.set_stream_handler(IDENTIFY_PROTOCOL_ID, identify_handler)

    async with (
        host_a.run(listen_addrs=listen_addrs),
        trio.open_nursery() as nursery,
    ):
        # Start the peer-store cleanup task
        nursery.start_soon(host_a.get_peerstore().start_cleanup_task, 60)

        # Get all available addresses with peer ID
        all_addrs = host_a.get_addrs()

        if use_varint_format:
            format_name = "length-prefixed"
            print(f"First host listening (using {format_name} format).")
            print("Listener ready, listening on:\n")
            for addr in all_addrs:
                print(f"{addr}")

            # Use optimal address for the client command
            optimal_addr = get_optimal_binding_address(port)
            optimal_addr_with_peer = (
                f"{optimal_addr}/p2p/{host_a.get_id().to_string()}"
            )
            print(
                f"\nRun this from the same folder in another console:\n\n"
                f"identify-demo -d {optimal_addr_with_peer}\n"
            )
            print("Waiting for incoming identify request...")
        else:
            format_name = "raw protobuf"
            print(f"First host listening (using {format_name} format).")
            print("Listener ready, listening on:\n")
            for addr in all_addrs:
                print(f"{addr}")

            # Use optimal address for the client command
            optimal_addr = get_optimal_binding_address(port)
            optimal_addr_with_peer = (
                f"{optimal_addr}/p2p/{host_a.get_id().to_string()}"
            )
            print(
                f"\nRun this from the same folder in another console:\n\n"
                f"identify-demo -d {optimal_addr_with_peer}\n"
            )
            print("Waiting for incoming identify request...")

        # Add a custom handler to show connection events
        async def custom_identify_handler(stream):
            peer_id = stream.muxed_conn.peer_id
            print(f"\n🔗 Received identify request from peer: {peer_id}")

            # Show remote address in multiaddr format
            try:
                from libp2p.identity.identify.identify import (
                    _remote_address_to_multiaddr,
                )

                remote_address = stream.get_remote_address()
                if remote_address:
                    observed_multiaddr = _remote_address_to_multiaddr(
                        remote_address
                    )
                    # Add the peer ID to create a complete multiaddr
                    complete_multiaddr = f"{observed_multiaddr}/p2p/{peer_id}"
                    print(f"   Remote address: {complete_multiaddr}")
                else:
                    print(f"   Remote address: {remote_address}")
            except Exception:
                print(f"   Remote address: {stream.get_remote_address()}")

            # Call the original handler
            await identify_handler(stream)

            print(f"✅ Successfully processed identify request from {peer_id}")

        # Replace the handler with our custom one
        host_a.set_stream_handler(IDENTIFY_PROTOCOL_ID, custom_identify_handler)

        try:
            await trio.sleep_forever()
        except KeyboardInterrupt:
            print("\n🛑 Shutting down listener...")
            logger.info("Listener interrupted by user")
            return
```

--------------------------------

### Install uv

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Install uv using pip or winget.

```bash
# Using pip
pip install uv

# Or using winget
winget install --id=astral-sh.uv
```

--------------------------------

### Dialer host setup

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify.html

Sets up the second host as a dialer, finds a free port if needed, gets available interfaces, and starts the host with a peerstore cleanup task.

```python
else:
        # Create second host (dialer)
        from libp2p.utils.address_validation import (
            find_free_port,
            get_available_interfaces,
            get_optimal_binding_address,
        )

        if port <= 0:
            port = find_free_port()

        listen_addrs = get_available_interfaces(port)
        host_b = new_host()

        async with (
            host_b.run(listen_addrs=listen_addrs),
            trio.open_nursery() as nursery,
        ):
            # Start the peer-store cleanup task
            nursery.start_soon(host_b.get_peerstore().start_cleanup_task, 60)

            # Connect to the first host
```

--------------------------------

### get_installed_packages()

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.observability.html

Gets installed packages.

```APIDOC
## get_installed_packages()

### Description
Gets installed packages.

### Method
Function

### Signature
`get_installed_packages()`

### Parameters
None

### Returns
- **list** - List of installed packages.
```

--------------------------------

### Listener/Dialer Example Header

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify_push.html

Starts the listener/dialer example script with a shebang and docstring explaining the identify/push protocol demonstration.

```python
#!/usr/bin/env python3
"""
Example demonstrating the identify/push protocol with separate listener and dialer
roles.

This example shows how to:
1. Set up a listener host with the identify/push protocol handler
2. Connect to the listener from a dialer peer
3. Push identify information to the listener
4. Receive and process identify/push messages

Usage:

```

--------------------------------

### examples.nat.listener.main()

Source: https://py-libp2p.readthedocs.io/en/stable/examples.nat.html

Parse arguments and run the listener node.

```APIDOC
## main()

### Description
Parse arguments and run the listener node.

### Function Signature
`main() -> None`

### Parameters
None

### Returns
- **None**
```

--------------------------------

### Python imports and logging setup for identify push example

Source: https://py-libp2p.readthedocs.io/en/stable/examples.identify_push.html

Imports required modules and configures logging for the identify push example. Sets logging levels to WARNING for multiaddr and libp2p, and creates a logger for the example.

```python
import argparse
import logging
import sys

import multiaddr
from multiaddr import (
    Multiaddr,
)
import trio

from libp2p import (
    new_host,
)
from libp2p.abc import (
    INetStream,
)
from libp2p.crypto.secp256k1 import (
    create_new_key_pair,
)
from libp2p.identity.identify import (
    ID as ID_IDENTIFY,
    identify_handler_for,
)
from libp2p.identity.identify.identify import (
    _remote_address_to_multiaddr,
)
from libp2p.identity.identify.pb.identify_pb2 import (
    Identify,
)
from libp2p.identity.identify_push import (
    ID_PUSH as ID_IDENTIFY_PUSH,
    identify_push_handler_for,
    push_identify_to_peer,
)
from libp2p.peer.peerinfo import (
    info_from_p2p_addr,
)

# Configure minimal logging
logging.basicConfig(level=logging.WARNING)
logging.getLogger("multiaddr").setLevel(logging.WARNING)
logging.getLogger("libp2p").setLevel(logging.WARNING)

# Configure logging
logger = logging.getLogger("libp2p.identity.identify-push-example")
```

--------------------------------

### examples.path_handling.path_handling.setup_logging(level: int = 20)

Source: https://py-libp2p.readthedocs.io/en/stable/examples.path_handling.html

Configures logging for the example. The level parameter is an integer representing the logging level (default 20 = INFO).

```APIDOC
## examples.path_handling.path_handling.setup_logging(level: int = 20)

### Description
Configure logging for the example.

### Method
Function

### Signature
`setup_logging(level: int = 20) -> None`

### Parameters
- **level** (int) - Optional - Logging level (default 20 = INFO)

### Returns
None
```

--------------------------------

### Start a rendezvous server on a specific port

Source: https://py-libp2p.readthedocs.io/en/stable/examples.rendezvous.html

Run the rendezvous example script in server mode on port 8080. This starts a rendezvous server that clients can connect to.

```bash
$ python rendezvous.py --mode server --port 8080
```

--------------------------------

### examples.path_handling.path_handling.main()

Source: https://py-libp2p.readthedocs.io/en/stable/examples.path_handling.html

Runs the robust path handling demo. This is the main entry point for the example script.

```APIDOC
## examples.path_handling.path_handling.main()

### Description
Runs the robust path handling demo.

### Method
Function

### Signature
`main() -> None`

### Parameters
None

### Returns
None
```

--------------------------------

### Set up development environment with uv

Source: https://py-libp2p.readthedocs.io/en/stable/contributing.html

Sync development dependencies and install pre-commit hooks using uv.

```bash
uv sync --group dev
uv run pre-commit install
```

--------------------------------

### Create virtual environment and install py-libp2p with uv

Source: https://py-libp2p.readthedocs.io/en/stable/install.html

Runs uv sync to create a .venv in the project directory and install the package.

```bash
uv sync
```

--------------------------------

### generate_self_signed_cert()

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.security.tls.html

Generates a self-signed certificate for testing purposes. This is a utility function based on the guide examples. Returns a tuple of (private key, certificate).

```APIDOC
## generate_self_signed_cert()

### Description
Generates a self-signed certificate for testing purposes. This is a utility function based on the guide examples.

### Returns
- **tuple[EllipticCurvePrivateKey, Certificate]** - Tuple of (private key, certificate)
```

--------------------------------

### examples.health_monitoring.configure.main()

Source: https://py-libp2p.readthedocs.io/en/stable/examples.health_monitoring.html

Main entry point for the health monitoring configuration example. It sets up the exporter port and runs the configuration.

```APIDOC
## Function: main

### Description
Main entry point for the health monitoring configuration example. It sets up the exporter port and runs the configuration.

### Signature
`main() -> None`

### Parameters
None

### Returns
None

### Example
```python
from examples.health_monitoring.configure import main
main()
```
```

--------------------------------

### examples.tls.example_tls_client.main

Source: https://py-libp2p.readthedocs.io/en/stable/examples.tls.html

Main entry point for the TLS client example.

```APIDOC
## examples.tls.example_tls_client.main

### Description
Main entry point for the TLS client example.

### Signature
```python
def main() -> None
```

### Returns
None

```

--------------------------------

### Run Bitswap examples in provider, client, and demo modes

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.bitswap.html

Commands to run the Bitswap example script in different modes. Provider mode serves blocks, client mode requests blocks from a provider, and demo mode runs an automated demonstration. The provider mode can optionally specify a file to serve.

```bash
# Provider
python examples/bitswap/bitswap.py --mode provider

# Provider with file
python examples/bitswap/bitswap.py --mode provider --file myfile.txt

# Client
python examples/bitswap/bitswap.py --mode client --provider <multiaddr>

# Demo
python examples/bitswap/bitswap.py --mode demo
```

--------------------------------

### PortDemultiplexer usage example

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.transport.html

Shows how to create a PortDemultiplexer, register listeners for TCP and WebSocket transports, and start listening. Must call demultiplexed_listen() before listen().

```python
port_demux = PortDemultiplexer("0.0.0.0", 4001)

# TCP transport registers for raw libp2p connections
tcp_listener = port_demux.demultiplexed_listen(
    maddr, DemultiplexedConnType.MULTISTREAM_SELECT
)

# WebSocket transport registers for HTTP-upgrade connections
ws_listener = port_demux.demultiplexed_listen(
    maddr, DemultiplexedConnType.HTTP
)

await port_demux.listen(maddr)   # binds the socket and starts routing
```

--------------------------------

### examples.nat.listener.setup_listener_node(port, relay_addr, seed=None)

Source: https://py-libp2p.readthedocs.io/en/stable/examples.nat.html

Set up and run a listener node behind NAT.

```APIDOC
## setup_listener_node(port, relay_addr, seed=None)

### Description
Set up and run a listener node behind NAT.

### Function Signature
`async setup_listener_node(port: int, relay_addr: str, seed: int | None = None) -> None`

### Parameters
- **port** (int) - Required - The port to listen on.
- **relay_addr** (str) - Required - The relay address to use for circuit relay.
- **seed** (int | None) - Optional - Seed for generating a fixed private key. Default is None.

### Returns
- **None**
```

--------------------------------

### connect(nursery: Nursery) -> None

Source: https://py-libp2p.readthedocs.io/en/stable/libp2p.transport.quic.html

Establishes the QUIC connection using a trio nursery for background tasks. Handles raw transport setup, TLS handshake, and peer identity verification. Does not call start() or set event_started.

```APIDOC
## connect(nursery: Nursery) -> None

### Description
Establishes the QUIC connection using a trio nursery for background tasks. This method handles all raw transport setup: physical connection initiation (for clients), background event-loop tasks, TLS handshake completion, and peer identity verification. It intentionally does NOT call start() or set event_started — those are exclusively owned by Swarm.add_conn() to maintain a uniform lifecycle contract across all transports.

### Method
Async function

### Signature
`async connect(nursery: Nursery) -> None`

### Parameters
- **nursery** (Nursery) - Required - Trio nursery for managing connection background tasks.

### Returns
None
```

--------------------------------

### Start a provider to share a file

Source: https://py-libp2p.readthedocs.io/en/stable/examples.bitswap.html

Run the Bitswap example in provider mode to share a file. The provider creates a Merkle DAG, stores blocks in memory, displays the root CID and connection command, and waits for client connections.

```bash
python examples/bitswap/bitswap.py --mode provider --file myfile.pdf
```

--------------------------------

### Perf Example Command Line Options

Source: https://py-libp2p.readthedocs.io/en/stable/examples.perf.html

Lists the command line options for the perf example script: -p for listening port (default random free port), -d for destination multiaddr (if not set, runs as server), -u for upload size in units of 256 bytes (default 10), -D for download size in units of 256 bytes (default 10).

```text
-p, --port          Listening port (default: random free port)
-d, --destination   Destination multiaddr (if not set, runs as server)
-u, --upload        Upload size in units of 256 bytes (default: 10)
-D, --download      Download size in units of 256 bytes (default: 10)
```

--------------------------------

### Setup Relay Node with Circuit Relay v2

Source: https://py-libp2p.readthedocs.io/en/stable/examples.circuit_relay.html

Configures and runs a relay node with all relay capabilities (HOP, STOP, CLIENT). Sets relay limits, initializes CircuitV2Protocol with allow_hop=True, registers stream handlers, and starts the relay service.

```python
async def setup_relay_node(port: int, seed: int | None = None) -> None:
    """Set up and run a relay node."""
    logger.info("Starting relay node...")

    # Create host with a fixed key if seed is provided
    key_pair = create_new_key_pair(generate_fixed_private_key(seed) if seed else None)
    logger.debug("[RELAY] created key_pair=%s", type(key_pair).__name__)
    host = new_host(key_pair=key_pair)
    logger.debug("[RELAY] host initialized | peer_id=%s", host.get_id())

    # Configure the relay
    limits = RelayLimits(
        duration=3600,  # 1 hour
        data=1024 * 1024 * 100,  # 100 MB
        max_circuit_conns=10,
        max_reservations=5,
    )

    relay_config = RelayConfig(
        roles=RelayRole.HOP | RelayRole.STOP | RelayRole.CLIENT,  # All capabilities
        limits=limits,
    )

    # Initialize the protocol
    protocol = CircuitV2Protocol(host, limits=limits, allow_hop=True)
    logger.debug(
        "[RELAY] CircuitV2Protocol initialized | allow_hop=%s |",
        "limits(duration=%s,data=%s,max_circuit_conns=%s,max_reservations=%s)",
        True,
        limits.duration,
        limits.data,
        limits.max_circuit_conns,
        limits.max_reservations,
    )

    # Start the host
    listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")

    async with host.run(listen_addrs=[listen_addr]):
        # Print information about this node
        peer_id = host.get_id()
        logger.info(f"Relay node started with ID: {peer_id}")

        addrs = host.get_addrs()
        for addr in addrs:
            logger.info(f"Listening on: {addr}")

        # Register protocol handlers
        logger.debug("[RELAY] registering stream handlers")
        host.set_stream_handler(EXAMPLE_PROTOCOL_ID, handle_example_protocol)
        host.set_stream_handler(PROTOCOL_ID, protocol._handle_hop_stream)
        host.set_stream_handler(STOP_PROTOCOL_ID, protocol._handle_stop_stream)
        logger.debug("[RELAY] protocol handlers registered")

        # Start the relay protocol service
        async with background_trio_service(protocol):
            logger.info("Circuit relay protocol started")

            # Create and register the transport
            CircuitV2Transport(host, protocol, relay_config)
            logger.info(
                "Circuit relay transport initialized | ",
                "enable_hop=%r enable_stop=%r enable_client=%r",
                relay_config.enable_hop,
                relay_config.enable_stop,
                relay_config.enable_client,
            )

            print("\nRelay node is running. Use the following address to connect:")
            print(f"{addrs[0]}")
            print("\nPress Ctrl+C to exit\n")

            # Keep the relay running
            await trio.sleep_forever()
```

--------------------------------

### Request/Response Demo with JSONCodec

Source: https://py-libp2p.readthedocs.io/en/stable/examples.request_response.html

Complete example for the libp2p request/response helper. Run once without -d to start a listener, then run again with -d to send a request. Uses JSONCodec and a custom protocol ID. The listener prints its multiaddr and waits; the dialer connects and sends a message, printing the response.

```python
from __future__ import annotations

import argparse
import logging
import random
import secrets

import multiaddr
import trio

from libp2p import new_host
from libp2p.crypto.secp256k1 import create_new_key_pair
from libp2p.custom_types import TProtocol
from libp2p.peer.peerinfo import info_from_p2p_addr
from libp2p.request_response import JSONCodec, RequestResponse
from libp2p.utils.address_validation import (
    find_free_port,
    get_available_interfaces,
    get_optimal_binding_address,
)

logging.basicConfig(level=logging.WARNING)
logging.getLogger("multiaddr").setLevel(logging.WARNING)
logging.getLogger("libp2p").setLevel(logging.WARNING)

PROTOCOL_ID = TProtocol("/example/request-response/1.0.0")


async def run(
    port: int,
    destination: str | None,
    message: str,
    seed: int | None = None,
) -> None:
    if port <= 0:
        port = find_free_port()
    listen_addr = get_available_interfaces(port)

    if seed is not None:
        random.seed(seed)
        secret_number = random.getrandbits(32 * 8)
        secret = secret_number.to_bytes(length=32, byteorder="big")
    else:
        secret = secrets.token_bytes(32)

    host = new_host(key_pair=create_new_key_pair(secret))
    rr = RequestResponse(host)
    codec = JSONCodec()

    async def handler(request: dict[str, str], context) -> dict[str, str]:
        return {
            "message": request["message"],
            "echo": request["message"].upper(),
            "peer": str(context.peer_id),
        }

    async with host.run(listen_addrs=listen_addr), trio.open_nursery() as nursery:
        nursery.start_soon(host.get_peerstore().start_cleanup_task, 60)
        print(f"I am {host.get_id().to_string()}")

        if not destination:
            rr.set_handler(PROTOCOL_ID, handler=handler, codec=codec)
            peer_id = host.get_id().to_string()
            print("Listener ready, listening on:\n")
            for addr in listen_addr:
                print(f"{addr}/p2p/{peer_id}")

            optimal_addr = get_optimal_binding_address(port)
            optimal_addr_with_peer = f"{optimal_addr}/p2p/{peer_id}"
            print(
                "\nRun this from the same folder in another console:\n\n"
                f"request-response-demo -d {optimal_addr_with_peer} --message hello\n"
            )
            print("Waiting for incoming requests...")
            await trio.sleep_forever()

        destination_str = destination
        if destination_str is None:
            raise ValueError("destination is required in dialer mode")
        maddr = multiaddr.Multiaddr(destination_str)
        info = info_from_p2p_addr(maddr)
        await host.connect(info)
        response = await rr.send_request(
            peer_id=info.peer_id,
            protocol_ids=[PROTOCOL_ID],
            request={"message": message},
            codec=codec,
        )
        print(f"Sent: {message}")
        print(f"Received: {response}")


def main() -> None:
    description = """
    Demonstrates the request/response helper with a single JSON request and response.
    Run once without -d to start a listener, then run again with -d to send a request.
    """
    example_maddr = (
        "/ip4/[HOST_IP]/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
    )
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument("-p", "--port", default=0, type=int, help="source port")
    parser.add_argument(
        "-d",
        "--destination",
        type=str,
        help=f"destination multiaddr string, e.g. {example_maddr}",
    )
    parser.add_argument(
        "--message",
        type=str,
        default="hello",
        help="JSON message payload to send",
    )
    parser.add_argument(
        "-s",
        "--seed",
        type=int,
        help="seed the RNG to make peer IDs reproducible",
    )
    args = parser.parse_args()
    try:
        trio.run(run, args.port, args.destination, args.message, args.seed)
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()
```

--------------------------------

### examples.multi_transport.client.main()

Source: https://py-libp2p.readthedocs.io/en/stable/examples.multi_transport.html

Entry point for the multi-transport echo client. Parses command-line arguments and runs the client.

```APIDOC
## examples.multi_transport.client.main()

### Description
Entry point for the multi-transport echo client. Parses command-line arguments and runs the client.

### Method
main()

### Parameters
None

### Returns
None

```
Tokens:
Raw