> For the complete documentation index, see [llms.txt](https://docs.hypertensor.org/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.hypertensor.org/copy-of-subnet-template/communication/p2p-protocol.md).

# P2P Protocol

## Introduction

Direct P2P communication that opens a stream between two peers can be achieved by building a custom P2P protocol.

To call a specific peer, its multiaddress is required. While all peers are connected through distributed algorithms using a Kademlia DHT (distributed hash table), not all peers are directly connected. In a P2P network, “connected” does not mean “directly connected.” It means **reachable through other peers**. Unlike with GossipSub, where you don't need a peer's multiaddress to send or receive a message, you do need it with direct P2P communication.

#### Acquiring A Multiaddress

A peer's multiaddress can be stored on-chain and fetched there (a node adding a multiaddress on-chain is optional, but can be required by subnets to allow them into the subnets P2P network); the subnet can set up a heartbeat system using a GossipSub that shares each peer's multiaddress where each peer stores that data and keeps it updated, or a peer can ask a directly connected peer for another peer's multiaddress from itself or its neighbors.

Once a multiaddress is acquired, then a peer can reach out to the desired peer.

## Building Custom Protocol

Building a custom P2P protocol for the subnets use case or use cases is achieved by inheriting the ProtocolBase template class.

#### Define the protobuf messages

```protobuf
syntax = "proto3";

package task_protocol.pb;

message TaskRequest {
  string task_id = 1;
  bytes payload = 2;
}

message TaskResponse {
  bool accepted = 1;
  bytes result = 2;
  string error = 3;
}
```

**Generate the Python protobuf module from the repository root:**

```bash
protoc --python_out=. --mypy_out=. subnet/protocols/pb/task_protocol.proto
```

This creates `subnet/protocols/pb/task_protocol_pb2.py` and, when `mypy-protobuf` is installed, `subnet/protocols/pb/task_protocol_pb2.pyi`. If you are not generating type stubs, omit `--mypy_out=.`. To regenerate it through the Makefile, add the proto path to the `PB` variable.

#### Create a protocol class

This is achieved by subclassing `ProtobufRequestResponseProtocolTemplate`:

```python
import logging

from libp2p.abc import IHost
from multiaddr import Multiaddr

from subnet.protocols.pb.task_protocol_pb2 import (
    TaskRequest,
    TaskResponse,
)
from subnet.protocols.protocol_base import (
    IncomingRequestContext,
    ProtobufRequestResponseProtocolTemplate,
)

logger = logging.getLogger("task_protocol/1.0.0")
PROTOCOL_ID = "/subnet/task_protocol/1.0.0"

"""
Example function that does a task when a peer is requested.

This can be a class or a function.
"""
async def run_task(payload: bytes) -> bytes:
    return payload.upper()


class TaskProtocol(
    ProtobufRequestResponseProtocolTemplate[TaskRequest, TaskResponse],
):
    def __init__(self, host: IHost) -> None:
        super().__init__(
            host=host,
            protocol_id=PROTOCOL_ID,
            request_message_type=TaskRequest,
            response_message_type=TaskResponse,
            log=logger,
        )

    async def call_remote(
        self,
        destination: Multiaddr | str,
        task_id: str,
        payload: bytes,
    ) -> TaskResponse:
        """Call out to a peer to perform a task"""
        request = TaskRequest(task_id=task_id, payload=payload)
        return await super().call_remote(destination, request)

    async def handle_request(
        self,
        context: IncomingRequestContext,
        request: TaskRequest,
    ) -> TaskResponse:
        """A peer is calling us to perform a task"""
        logger.info(
            "Received task %s from %s",
            request.task_id,
            context.peer_id,
        )
        result = await run_task(request.payload)
        return TaskResponse(accepted=True, result=result)
```

#### Instantiate the protocol

The base class registers the stream handler automatically.

```
task_protocol = TaskProtocol(host)
response = await task_protocol.call_remote(
    remote_multiaddr,
    task_id="score-batch-001",
    payload=b"...",
)
```