Network API

The network API module provides a template for spanning a REST API that acts as a bridge between your core P2P Node and separate local or remote processes.

It is designed for cases where you might have heavy workloads, such as AI inference tasks, external logic servers, or monitoring dashboards, that operate outside of the node but still need to invoke P2P protocols, query peer statuses, or publish consensus/gossip messages.

The Network API enables you to call an API endpoint hosted on the P2P process from a separate local or remote process to access and interact with the P2P layer of your peer.

For example, external processes can call this API endpoint on the P2P process to trigger a gossipsub publish, call the server to retrieve data from the database, call a peer, or call to execute on a blockchain extrinsic. This can also be used to fully compartmentalize the node's business logic from the P2P layer, where that separate process can trigger all communication between peers.

Network API + P2P Communication

The Network API can also be used to port a previously developed distributed AI application that primarily uses an API to communicate between servers and turn them into a decentralized AI application by communicating with peers using the GossipSub or the API Protocol.

The Network API can effectively be used as a middleman between the previously built application and the new P2P layer, where it can safely communicate with peers using built-in encrypted messaging.

---

Configuration

Similar to the API Protocol, it's suggested to use a JSON file for the Network APIs configuration so it can be updated without restarting the peer and reconnecting to the network.

JSON

{
  "host": "127.0.0.1",
  "port": 8000,
  "whitelist_ips": [
    "127.0.0.1",
    "192.168.1.100"
  ],
  "enable_api": true
}

• listen_host: Interface address the server binds to, such as 127.0.0.1 or 0.0.0.0.

• port: TCP port the server listens on.

• whitelist_ips: Client IPs allowed to call the API; an empty list allows all IPs.

• enable_api: Whether the API server should start.

Configuration Class

class ApiConfig(BaseModel):
    listen_host: str = "127.0.0.1"
    port: int = 8000
    whitelist_ips: List[str] = ["127.0.0.1"]
    enable_api: bool = True

Initiate the Server

The NetworkApi class can be initialized with everything from the peer host, DHT, gossipsub, pubsub, database, etc., allowing access to anything in the P2P layer from another process.

network_api = NetworkApi(...)
api_server = NetworkApiServer(network_api, config="subnet/network_api/config.json")
await api_server.start()

Add Custom Logic

Within the NetworkApi class, add functions that the API can call to access operations or data.

In the following example, the API can be called from a separate process to publish a message to the pubsub for all other peers to receive.

class NetworkApi:
    ...
    async def publish_topic(self, topic: str, message: bytes):
        """
        Publish a message to a topic.
        """
        if self.pubsub is None:
            raise Exception("Pubsub not initialized")
        try:
            await self.pubsub.publish(topic, message)
        except Exception as e:
            logger.error(f"Failed to publish message to topic {topic}: {e}")
            raise
            
    async def call_api_protocol(
        self,
        destination: Multiaddr,
        route: str,
        method: str = "GET",
        headers: dict = None,
        body: bytes = b"",
    ) -> bytes:
        """
        Call a remote peer's API and wait for a single unary response.
        """
        if self.api_protocol is None:
            raise Exception("API protocol not initialized")
        try:
            return await self.api_protocol.call_remote(destination, route, method, headers, body)
        except Exception as e:
            logger.error(f"Failed to call API protocol on peer {destination}: {e}")
            raise

Customize Endpoints

By default, the template includes a single demonstration route: [POST] /v1/publish that expects a topic and message payload (this is just an example endpoint).

To expand this to handle custom requirements (like pinging specific tasks or triggering DHT lookups), modify the _setup_template_routes() method in network_api.py.

class NetworkApiServer:
    ...
    def _setup_template_routes(self):
    
        class PublishRequest(BaseModel):
            topic: str
            message: str

        @self.router.post("/v1/my-custom-task")
        async def run_my_task(request: PublishRequest):
            await self.network_api.publish_topic(request.topic, request.message.encode("utf-8"))
            
        class ApiProtocolRequest(BaseModel):
            destination: Multiaddr
            route: str

        @self.router.post("/v1/api-protocol")
        async def run_call_api_protocol(request: ApiProtocolRequest):
            await self.network_api.call_api_protocol(request.destination, request.route)

Alternatively, inject routes dynamically from another location dynamically:

async def my_external_handler():
    return "Handled!"

api_server.register_route(
    path="/v1/external", 
    endpoint=my_external_handler, 
    methods=["GET"]
)

Stopping Operations

Always ensure await api_server.stop() is captured on process teardown to prevent Uvicorn from abruptly terminating long-running external hook executions.