Server & Application

There are two main templates for deploying the application and P2P logic. The ServerBase and the ApplicationBase. With these base templates, you will build a Server and an Application class that inherits the bases. The server class you create will deploy the application class.

The server base automatically handles all peer discovery and has options to deploy the GossipSub.

Create Application

This is the logic the peer deploys and can be role-dependentrole dependant based on the needs of the subnet.

Example

class MyApplication(ApplicationBase):
    def __init__(
        self,
        *,
        role: "trainer" | "validator",
        key_pair: KeyPair,
        db: RocksDB,
        subnet_id: int,
        subnet_node_id: int,
        is_bootstrap: bool,
        telemetry: Telemetry | None = None,
    ) -> None:
        super().__init__()
        self.role= role
        self.key_pair = key_pair
        self.db = db
        self.subnet_id = subnet_id
        self.subnet_node_id = subnet_node_id
        self.is_bootstrap = is_bootstrap
        self.telemetry = telemetry

        async def setup(self, context: P2PNetworkContext) -> None:
            """
            Called once after the host, DHT, and optional pubsub are ready.
                
            The custom P2P logic should deploy from here, such as the custom P2P
            protocols, the API protocol. As well as non-application-based logic,
            such as the Network API.
            """
    
        async def start_application(self, context: P2PNetworkContext) -> None:
            """
            Called after the bootstrap and optional connection maintenance startup.
            
            Deploy all of the application logic for the peer
            """
    
        async def cleanup(self, context: P2PNetworkContext) -> None:
            """
            Called once while the network context is still available.
            
            Cleanup any processes created.
            """

Create Server

The server will initialize all of the P2P logic, such as peer discovery, and deploy the application logic

Example

class MyServer(ServerBase):
    def __init__(
        self,
        *,
        ip: str | None = None,
        port: int,
        bootstrap_addrs: Sequence[str] | None = None,
        key_pair: KeyPair,
        db: RocksDB,
        subnet_id: int,
        subnet_slot: int = 3,
        subnet_node_id: int,
        hypertensor: Hypertensor | LocalMockHypertensor,
        is_bootstrap: bool = False,
        telemetry: Telemetry | None = None,
        enable_mDNS: bool = False,
        enable_upnp: bool = False,
        enable_autotls: bool = False,
        resource_manager: ResourceManager | None = None,
        psk: str | None = None,
    ) -> None:
        application = MyApplication(
            role="trainer",
            key_pair=key_pair,
            db=db,
            subnet_id=subnet_id,
            subnet_node_id=subnet_node_id,
            is_bootstrap=is_bootstrap,
            telemetry=telemetry,
        )

        super().__init__(
            ip=ip or "0.0.0.0",
            port=port,
            application=application,
            key_pair=key_pair,
            bootstrap_addrs=bootstrap_addrs,
            use_available_interfaces=True,
            enable_pubsub=True,
            enable_random_walk=True,
            enable_mDNS=enable_mDNS,
            enable_upnp=enable_upnp,
            enable_autotls=enable_autotls,
            resource_manager=resource_manager,
            psk=psk,
            max_connections_per_peer=6,
            enable_proof_of_stake=True,
            db=db,
            subnet_id=subnet_id,
            subnet_slot=subnet_slot,
            subnet_node_id=subnet_node_id,
            hypertensor=hypertensor,
            is_bootstrap=is_bootstrap,
            enable_subnet_info_tracker=True,
            enable_connection_maintenance=True,
            strict_maintain_connections=True,
            telemetry=telemetry,
            maintain_connections_log_level=logging.DEBUG,
        )