Api stubs (#19979)

* harvester_api_protocol

* move to apis

* Remove `chia.apis` => `chia.harvester` dependency.

* Remove circular dependencies between `chia.apis` and service modules.

* SolverAPIStub, and add solver to tach.toml.

* Make the stubs (lack of) implementations more consistent.

* Use `stub_metadata_for_type` instead of `class_for_type`.

The only thing ever used from `ApiProtocol` in `WSConnection`
was the metadata field, so let's pass along only that.

* NewSignagePointHarvester2

* remove some stuff

chia/_tests/connection_utils.py

@@ -12,7 +12,7 @@ from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
 
 from chia._tests.util.time_out_assert import time_out_assert
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.protocols.outbound_message import NodeType
 from chia.protocols.shared_protocol import default_capabilities
 from chia.server.server import ChiaServer, ssl_context_for_client
@@ -98,7 +98,7 @@ async def add_dummy_connection_wsc(
         100,
         30,
         local_capabilities_for_handshake=default_capabilities[type] + additional_capabilities,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
     await wsc.perform_handshake(server._network_id, dummy_port, type)
     if wsc.incoming_message_task is not None:

chia/_tests/core/ssl/test_ssl.py

@@ -9,7 +9,7 @@ import aiohttp
 import pytest
 from chia_rs.sized_bytes import bytes32
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.protocols.outbound_message import NodeType
 from chia.protocols.shared_protocol import default_capabilities
 from chia.server.server import ChiaServer, ssl_context_for_client
@@ -38,7 +38,7 @@ async def establish_connection(server: ChiaServer, self_hostname: str, ssl_conte
             100,
             30,
             local_capabilities_for_handshake=default_capabilities[NodeType.FULL_NODE],
-            class_for_type=ApiProtocolRegistry,
+            stub_metadata_for_type=StubMetadataRegistry,
         )
         await wsc.perform_handshake(server._network_id, dummy_port, NodeType.FULL_NODE)
         await wsc.close()

chia/apis.py

@@ -1,21 +0,0 @@
-from __future__ import annotations
-
-from chia.farmer.farmer_api import FarmerAPI
-from chia.full_node.full_node_api import FullNodeAPI
-from chia.harvester.harvester_api import HarvesterAPI
-from chia.introducer.introducer_api import IntroducerAPI
-from chia.protocols.outbound_message import NodeType
-from chia.server.api_protocol import ApiProtocol
-from chia.solver.solver_api import SolverAPI
-from chia.timelord.timelord_api import TimelordAPI
-from chia.wallet.wallet_node_api import WalletNodeAPI
-
-ApiProtocolRegistry: dict[NodeType, type[ApiProtocol]] = {
-    NodeType.FULL_NODE: FullNodeAPI,
-    NodeType.WALLET: WalletNodeAPI,
-    NodeType.INTRODUCER: IntroducerAPI,
-    NodeType.TIMELORD: TimelordAPI,
-    NodeType.FARMER: FarmerAPI,
-    NodeType.HARVESTER: HarvesterAPI,
-    NodeType.SOLVER: SolverAPI,
-}

chia/apis/__init__.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from chia.apis.farmer_stub import FarmerApiStub
+from chia.apis.full_node_stub import FullNodeApiStub
+from chia.apis.harvester_stub import HarvesterApiStub
+from chia.apis.introducer_stub import IntroducerApiStub
+from chia.apis.solver_stub import SolverApiStub
+from chia.apis.stub_protocol_registry import StubMetadataRegistry
+from chia.apis.timelord_stub import TimelordApiStub
+from chia.apis.wallet_stub import WalletNodeApiStub
+
+__all__ = [
+    "FarmerApiStub",
+    "FullNodeApiStub",
+    "HarvesterApiStub",
+    "IntroducerApiStub",
+    "SolverApiStub",
+    "StubMetadataRegistry",
+    "TimelordApiStub",
+    "WalletNodeApiStub",
+]

chia/apis/farmer_stub.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+from typing_extensions import Protocol
+
+from chia.protocols import farmer_protocol, harvester_protocol
+from chia.protocols.harvester_protocol import (
+    PlotSyncDone,
+    PlotSyncPathList,
+    PlotSyncPlotList,
+    PlotSyncStart,
+    RespondPlots,
+    RespondSignatures,
+)
+from chia.protocols.outbound_message import Message
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class FarmerApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for FarmerAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual FarmerAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    # Create a concrete instance for decorators while keeping the ClassVar type hint for mypy
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the farmer is ready."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_proof_of_space(
+        self, new_proof_of_space: harvester_protocol.NewProofOfSpace, peer: WSChiaConnection
+    ) -> None:
+        """Handle new proof of space from harvester."""
+        ...
+
+    @metadata.request()
+    async def respond_signatures(self, response: RespondSignatures) -> None:
+        """Handle signature response from harvester."""
+        ...
+
+    @metadata.request()
+    async def new_signage_point(self, new_signage_point: farmer_protocol.NewSignagePoint) -> None:
+        """Handle new signage point from full node."""
+        ...
+
+    @metadata.request()
+    async def request_signed_values(self, full_node_request: farmer_protocol.RequestSignedValues) -> Optional[Message]:
+        """Handle request for signed values from full node."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def farming_info(self, request: farmer_protocol.FarmingInfo, peer: WSChiaConnection) -> None:
+        """Handle farming info from full node."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_plots(self, _: RespondPlots, peer: WSChiaConnection) -> None:
+        """Handle respond plots from harvester."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_start(self, message: PlotSyncStart, peer: WSChiaConnection) -> None:
+        """Handle plot sync start."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_loaded(self, message: PlotSyncPlotList, peer: WSChiaConnection) -> None:
+        """Handle plot sync loaded."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_removed(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None:
+        """Handle plot sync removed."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_invalid(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None:
+        """Handle plot sync invalid."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_keys_missing(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None:
+        """Handle plot sync keys missing."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_duplicates(self, message: PlotSyncPathList, peer: WSChiaConnection) -> None:
+        """Handle plot sync duplicates."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def plot_sync_done(self, message: PlotSyncDone, peer: WSChiaConnection) -> None:
+        """Handle plot sync done."""
+        ...

chia/apis/full_node_stub.py

@@ -0,0 +1,380 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+from typing_extensions import Protocol
+
+from chia.protocols import (
+    farmer_protocol,
+    full_node_protocol,
+    introducer_protocol,
+    timelord_protocol,
+    wallet_protocol,
+)
+from chia.protocols.outbound_message import Message
+from chia.protocols.protocol_message_types import ProtocolMessageTypes
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class FullNodeApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for FullNodeAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual FullNodeAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the full node is ready."""
+        ...
+
+    # PEER PROTOCOL
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_peers])
+    async def request_peers(
+        self, _request: full_node_protocol.RequestPeers, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle peer request."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_peers(
+        self, request: full_node_protocol.RespondPeers, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle peers response."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_peers_introducer(
+        self, request: introducer_protocol.RespondPeersIntroducer, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle peers response from introducer."""
+        ...
+
+    # FULL NODE PROTOCOL
+    @metadata.request(peer_required=True, execute_task=True)
+    async def new_peak(self, request: full_node_protocol.NewPeak, peer: WSChiaConnection) -> None:
+        """Handle new peak from peer."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_transaction(
+        self, transaction: full_node_protocol.NewTransaction, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new transaction from peer."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_transaction])
+    async def request_transaction(self, request: full_node_protocol.RequestTransaction) -> Optional[Message]:
+        """Handle transaction request."""
+        ...
+
+    @metadata.request(peer_required=True, bytes_required=True)
+    async def respond_transaction(
+        self,
+        tx: full_node_protocol.RespondTransaction,
+        peer: WSChiaConnection,
+        tx_bytes: bytes = b"",
+        test: bool = False,
+    ) -> Optional[Message]:
+        """Handle transaction response from peer."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_proof_of_weight])
+    async def request_proof_of_weight(self, request: full_node_protocol.RequestProofOfWeight) -> Optional[Message]:
+        """Handle proof of weight request."""
+        ...
+
+    @metadata.request()
+    async def respond_proof_of_weight(self, request: full_node_protocol.RespondProofOfWeight) -> Optional[Message]:
+        """Handle proof of weight response."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_block, ProtocolMessageTypes.reject_block])
+    async def request_block(self, request: full_node_protocol.RequestBlock) -> Optional[Message]:
+        """Handle block request."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_blocks, ProtocolMessageTypes.reject_blocks])
+    async def request_blocks(self, request: full_node_protocol.RequestBlocks) -> Optional[Message]:
+        """Handle blocks request."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def reject_block(
+        self,
+        request: full_node_protocol.RejectBlock,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle block rejection."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def reject_blocks(
+        self,
+        request: full_node_protocol.RejectBlocks,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle blocks rejection."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_blocks(
+        self,
+        request: full_node_protocol.RespondBlocks,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle blocks response."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_block(
+        self,
+        respond_block: full_node_protocol.RespondBlock,
+        peer: WSChiaConnection,
+    ) -> Optional[Message]:
+        """Handle block response."""
+        ...
+
+    @metadata.request()
+    async def new_unfinished_block(
+        self, new_unfinished_block: full_node_protocol.NewUnfinishedBlock
+    ) -> Optional[Message]:
+        """Handle new unfinished block."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_unfinished_block])
+    async def request_unfinished_block(
+        self, request_unfinished_block: full_node_protocol.RequestUnfinishedBlock
+    ) -> Optional[Message]:
+        """Handle unfinished block request."""
+        ...
+
+    @metadata.request()
+    async def new_unfinished_block2(
+        self, new_unfinished_block: full_node_protocol.NewUnfinishedBlock2
+    ) -> Optional[Message]:
+        """Handle new unfinished block v2."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_unfinished_block])
+    async def request_unfinished_block2(
+        self, request_unfinished_block: full_node_protocol.RequestUnfinishedBlock2
+    ) -> Optional[Message]:
+        """Handle unfinished block v2 request."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_unfinished_block(
+        self,
+        respond_unfinished_block: full_node_protocol.RespondUnfinishedBlock,
+        peer: WSChiaConnection,
+    ) -> Optional[Message]:
+        """Handle unfinished block response."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_signage_point_or_end_of_sub_slot(
+        self, new_sp: full_node_protocol.NewSignagePointOrEndOfSubSlot, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new signage point or end of sub slot."""
+        ...
+
+    @metadata.request(
+        reply_types=[ProtocolMessageTypes.respond_signage_point, ProtocolMessageTypes.respond_end_of_sub_slot]
+    )
+    async def request_signage_point_or_end_of_sub_slot(
+        self, request: full_node_protocol.RequestSignagePointOrEndOfSubSlot
+    ) -> Optional[Message]:
+        """Handle signage point or end of sub slot request."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_signage_point(
+        self, request: full_node_protocol.RespondSignagePoint, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle signage point response."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_end_of_sub_slot(
+        self, request: full_node_protocol.RespondEndOfSubSlot, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle end of sub slot response."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def request_mempool_transactions(
+        self,
+        request: full_node_protocol.RequestMempoolTransactions,
+        peer: WSChiaConnection,
+    ) -> Optional[Message]:
+        """Handle mempool transactions request."""
+        ...
+
+    # FARMER PROTOCOL
+    @metadata.request(peer_required=True)
+    async def declare_proof_of_space(
+        self, request: farmer_protocol.DeclareProofOfSpace, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle proof of space declaration from farmer."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def signed_values(
+        self, farmer_request: farmer_protocol.SignedValues, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle signed values from farmer."""
+        ...
+
+    # TIMELORD PROTOCOL
+    @metadata.request(peer_required=True)
+    async def new_infusion_point_vdf(
+        self, request: timelord_protocol.NewInfusionPointVDF, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new infusion point VDF from timelord."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_signage_point_vdf(
+        self, request: timelord_protocol.NewSignagePointVDF, peer: WSChiaConnection
+    ) -> None:
+        """Handle new signage point VDF from timelord."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_end_of_sub_slot_vdf(
+        self, request: timelord_protocol.NewEndOfSubSlotVDF, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new end of sub slot VDF from timelord."""
+        ...
+
+    @metadata.request(bytes_required=True, execute_task=True)
+    async def respond_compact_proof_of_time(
+        self, request: timelord_protocol.RespondCompactProofOfTime, request_bytes: bytes = b""
+    ) -> None:
+        """Handle compact proof of time response from timelord."""
+
+    @metadata.request(peer_required=True, bytes_required=True, execute_task=True)
+    async def new_compact_vdf(
+        self, request: full_node_protocol.NewCompactVDF, peer: WSChiaConnection, request_bytes: bytes = b""
+    ) -> None:
+        """Handle new compact VDF."""
+        ...
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_compact_vdf])
+    async def request_compact_vdf(self, request: full_node_protocol.RequestCompactVDF, peer: WSChiaConnection) -> None:
+        """Handle compact VDF request."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_compact_vdf(self, request: full_node_protocol.RespondCompactVDF, peer: WSChiaConnection) -> None:
+        """Handle compact VDF response."""
+        ...
+
+    # WALLET PROTOCOL
+    @metadata.request()
+    async def request_block_header(self, request: wallet_protocol.RequestBlockHeader) -> Optional[Message]:
+        """Handle block header request from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_additions(self, request: wallet_protocol.RequestAdditions) -> Optional[Message]:
+        """Handle additions request from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_removals(self, request: wallet_protocol.RequestRemovals) -> Optional[Message]:
+        """Handle removals request from wallet."""
+        ...
+
+    @metadata.request()
+    async def send_transaction(
+        self, request: wallet_protocol.SendTransaction, *, test: bool = False
+    ) -> Optional[Message]:
+        """Handle transaction send from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_puzzle_solution(self, request: wallet_protocol.RequestPuzzleSolution) -> Optional[Message]:
+        """Handle puzzle solution request from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_block_headers(self, request: wallet_protocol.RequestBlockHeaders) -> Optional[Message]:
+        """Handle block headers request from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_header_blocks(self, request: wallet_protocol.RequestHeaderBlocks) -> Optional[Message]:
+        """Handle header blocks request from wallet (deprecated)."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def register_for_ph_updates(
+        self, request: wallet_protocol.RegisterForPhUpdates, peer: WSChiaConnection
+    ) -> Message:
+        """Handle puzzle hash updates registration from wallet."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def register_for_coin_updates(
+        self, request: wallet_protocol.RegisterForCoinUpdates, peer: WSChiaConnection
+    ) -> Message:
+        """Handle coin updates registration from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_children(self, request: wallet_protocol.RequestChildren) -> Optional[Message]:
+        """Handle children request from wallet."""
+        ...
+
+    @metadata.request()
+    async def request_ses_hashes(self, request: wallet_protocol.RequestSESInfo) -> Message:
+        """Handle SES hashes request from wallet."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_fee_estimates])
+    async def request_fee_estimates(self, request: wallet_protocol.RequestFeeEstimates) -> Message:
+        """Handle fee estimates request from wallet."""
+        ...
+
+    @metadata.request(
+        peer_required=True,
+        reply_types=[ProtocolMessageTypes.respond_remove_puzzle_subscriptions],
+    )
+    async def request_remove_puzzle_subscriptions(
+        self, request: wallet_protocol.RequestRemovePuzzleSubscriptions, peer: WSChiaConnection
+    ) -> Message:
+        """Handle remove puzzle subscriptions request from wallet."""
+        ...
+
+    @metadata.request(
+        peer_required=True,
+        reply_types=[ProtocolMessageTypes.respond_remove_coin_subscriptions],
+    )
+    async def request_remove_coin_subscriptions(
+        self, request: wallet_protocol.RequestRemoveCoinSubscriptions, peer: WSChiaConnection
+    ) -> Message:
+        """Handle remove coin subscriptions request from wallet."""
+        ...
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_puzzle_state])
+    async def request_puzzle_state(
+        self, request: wallet_protocol.RequestPuzzleState, peer: WSChiaConnection
+    ) -> Message:
+        """Handle puzzle state request from wallet."""
+        ...
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_coin_state])
+    async def request_coin_state(self, request: wallet_protocol.RequestCoinState, peer: WSChiaConnection) -> Message:
+        """Handle coin state request from wallet."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_cost_info])
+    async def request_cost_info(self, _request: wallet_protocol.RequestCostInfo) -> Optional[Message]:
+        """Handle cost info request from wallet."""
+        ...

chia/apis/harvester_stub.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+from typing_extensions import Protocol
+
+from chia.protocols import harvester_protocol
+from chia.protocols.harvester_protocol import PlotSyncResponse
+from chia.protocols.outbound_message import Message
+from chia.protocols.protocol_message_types import ProtocolMessageTypes
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class HarvesterApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for HarvesterAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual HarvesterAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the harvester is ready."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def harvester_handshake(
+        self, harvester_handshake: harvester_protocol.HarvesterHandshake, peer: WSChiaConnection
+    ) -> None:
+        """Handshake between the harvester and farmer."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def new_signage_point_harvester(
+        self, new_challenge: harvester_protocol.NewSignagePointHarvester2, peer: WSChiaConnection
+    ) -> None:
+        """Handle new signage point from farmer."""
+        ...
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_signatures])
+    async def request_signatures(self, request: harvester_protocol.RequestSignatures) -> Optional[Message]:
+        """Handle signature request from farmer."""
+        ...
+
+    @metadata.request()
+    async def request_plots(self, _: harvester_protocol.RequestPlots) -> Message:
+        """Handle request for plot information."""
+        ...
+
+    @metadata.request()
+    async def plot_sync_response(self, response: PlotSyncResponse) -> None:
+        """Handle plot sync response."""
+        ...

chia/apis/introducer_stub.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+from typing_extensions import Protocol
+
+from chia.protocols.introducer_protocol import RequestPeersIntroducer
+from chia.protocols.outbound_message import Message
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class IntroducerApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for IntroducerAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual IntroducerAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the introducer is ready."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def request_peers_introducer(
+        self,
+        request: RequestPeersIntroducer,
+        peer: WSChiaConnection,
+    ) -> Optional[Message]:
+        """Handle request for peers from a node."""
+        ...

chia/apis/solver_stub.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar, Optional
+
+from typing_extensions import Protocol
+
+from chia.protocols.outbound_message import Message
+from chia.protocols.solver_protocol import SolverInfo
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+
+
+class SolverApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for SolverAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual SolverAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the solver is ready."""
+        ...
+
+    @metadata.request(peer_required=False)
+    async def solve(self, request: SolverInfo) -> Optional[Message]:
+        """Handle solver request."""
+        ...

chia/apis/stub_protocol_registry.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from chia.apis.farmer_stub import FarmerApiStub
+from chia.apis.full_node_stub import FullNodeApiStub
+from chia.apis.harvester_stub import HarvesterApiStub
+from chia.apis.introducer_stub import IntroducerApiStub
+from chia.apis.solver_stub import SolverApiStub
+from chia.apis.timelord_stub import TimelordApiStub
+from chia.apis.wallet_stub import WalletNodeApiStub
+from chia.protocols.outbound_message import NodeType
+from chia.server.api_protocol import ApiMetadata
+
+StubMetadataRegistry: dict[NodeType, ApiMetadata] = {
+    NodeType.FULL_NODE: FullNodeApiStub.metadata,
+    NodeType.WALLET: WalletNodeApiStub.metadata,
+    NodeType.INTRODUCER: IntroducerApiStub.metadata,
+    NodeType.TIMELORD: TimelordApiStub.metadata,
+    NodeType.FARMER: FarmerApiStub.metadata,
+    NodeType.HARVESTER: HarvesterApiStub.metadata,
+    NodeType.SOLVER: SolverApiStub.metadata,
+}

chia/apis/timelord_stub.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar
+
+from typing_extensions import Protocol
+
+from chia.protocols.timelord_protocol import NewPeakTimelord, NewUnfinishedBlockTimelord, RequestCompactProofOfTime
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+
+
+class TimelordApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for TimelordAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual TimelordAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the timelord is ready."""
+        ...
+
+    @metadata.request()
+    async def new_peak_timelord(self, new_peak: NewPeakTimelord) -> None:
+        """Handle new peak from full node."""
+        ...
+
+    @metadata.request()
+    async def new_unfinished_block_timelord(self, new_unfinished_block: NewUnfinishedBlockTimelord) -> None:
+        """Handle new unfinished block from full node."""
+        ...
+
+    @metadata.request()
+    async def request_compact_proof_of_time(self, vdf_info: RequestCompactProofOfTime) -> None:
+        """Handle request for compact proof of time."""
+        ...

chia/apis/wallet_stub.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import logging
+from typing import ClassVar
+
+from chia_rs import RespondToPhUpdates
+from typing_extensions import Protocol
+
+from chia.protocols import full_node_protocol, introducer_protocol, wallet_protocol
+from chia.protocols.wallet_protocol import (
+    CoinStateUpdate,
+    RejectAdditionsRequest,
+    RejectBlockHeaders,
+    RejectHeaderBlocks,
+    RejectHeaderRequest,
+    RejectPuzzleSolution,
+    RejectRemovalsRequest,
+    RespondAdditions,
+    RespondBlockHeader,
+    RespondBlockHeaders,
+    RespondChildren,
+    RespondHeaderBlocks,
+    RespondPuzzleSolution,
+    RespondRemovals,
+    RespondSESInfo,
+    RespondToCoinUpdates,
+    TransactionAck,
+)
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
+from chia.server.ws_connection import WSChiaConnection
+
+
+class WalletNodeApiStub(ApiProtocol, Protocol):
+    """Non-functional API stub for WalletNodeAPI
+
+    This is a protocol definition only - methods are not implemented and should
+    never be called. Use the actual WalletNodeAPI implementation at runtime.
+    """
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the wallet is ready."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_removals(self, response: RespondRemovals, peer: WSChiaConnection) -> None:
+        """Handle removals response from full node."""
+        ...
+
+    async def reject_removals_request(self, response: RejectRemovalsRequest, peer: WSChiaConnection) -> None:
+        """Handle reject removals request from full node."""
+        ...
+
+    @metadata.request()
+    async def reject_additions_request(self, response: RejectAdditionsRequest) -> None:
+        """Handle reject additions request from full node."""
+        ...
+
+    @metadata.request(peer_required=True, execute_task=True)
+    async def new_peak_wallet(self, peak: wallet_protocol.NewPeakWallet, peer: WSChiaConnection) -> None:
+        """Handle new peak from full node."""
+        ...
+
+    @metadata.request()
+    async def reject_header_request(self, response: RejectHeaderRequest) -> None:
+        """Handle reject header request from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_block_header(self, response: RespondBlockHeader) -> None:
+        """Handle block header response from full node."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_additions(self, response: RespondAdditions, peer: WSChiaConnection) -> None:
+        """Handle additions response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_proof_of_weight(self, response: full_node_protocol.RespondProofOfWeight) -> None:
+        """Handle proof of weight response from full node."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def transaction_ack(self, ack: TransactionAck, peer: WSChiaConnection) -> None:
+        """Handle transaction acknowledgment from full node."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_peers_introducer(
+        self, request: introducer_protocol.RespondPeersIntroducer, peer: WSChiaConnection
+    ) -> None:
+        """Handle peers response from introducer."""
+        ...
+
+    @metadata.request(peer_required=True)
+    async def respond_peers(self, request: full_node_protocol.RespondPeers, peer: WSChiaConnection) -> None:
+        """Handle peers response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_puzzle_solution(self, request: RespondPuzzleSolution) -> None:
+        """Handle puzzle solution response from full node."""
+        ...
+
+    @metadata.request()
+    async def reject_puzzle_solution(self, request: RejectPuzzleSolution) -> None:
+        """Handle reject puzzle solution from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_header_blocks(self, request: RespondHeaderBlocks) -> None:
+        """Handle header blocks response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_block_headers(self, request: RespondBlockHeaders) -> None:
+        """Handle block headers response from full node."""
+        ...
+
+    @metadata.request()
+    async def reject_header_blocks(self, request: RejectHeaderBlocks) -> None:
+        """Handle reject header blocks from full node."""
+        ...
+
+    @metadata.request()
+    async def reject_block_headers(self, request: RejectBlockHeaders) -> None:
+        """Handle reject block headers from full node."""
+        ...
+
+    @metadata.request(peer_required=True, execute_task=True)
+    async def coin_state_update(self, request: CoinStateUpdate, peer: WSChiaConnection) -> None:
+        """Handle coin state update from full node."""
+        ...
+
+    @metadata.request()  # type: ignore[type-var]
+    async def respond_to_ph_updates(self, request: RespondToPhUpdates) -> None:
+        """Handle puzzle hash updates response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_to_coin_updates(self, request: RespondToCoinUpdates) -> None:
+        """Handle coin updates response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_children(self, request: RespondChildren) -> None:
+        """Handle children response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_ses_hashes(self, request: RespondSESInfo) -> None:
+        """Handle SES hashes response from full node."""
+        ...
+
+    @metadata.request()
+    async def respond_blocks(self, request: full_node_protocol.RespondBlocks) -> None:
+        """Handle blocks response from full node."""
+        ...

chia/data_layer/start_data_layer.py

@@ -8,7 +8,7 @@ from typing import Any, Optional, cast
 
 from chia_rs.sized_ints import uint16
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.data_layer.data_layer import DataLayer
 from chia.data_layer.data_layer_api import DataLayerAPI
 from chia.data_layer.data_layer_rpc_api import DataLayerRpcApi
@@ -89,7 +89,7 @@ def create_data_layer_service(
         max_request_body_size=service_config.get("rpc_server_max_request_body_size", 26214400),
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/farmer/farmer_api.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 import logging
 import time
-from typing import TYPE_CHECKING, Any, ClassVar, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, Union
 
 import aiohttp
 from chia_rs import AugSchemeMPL, G2Element, PlotParam, PoolTarget, PrivateKey, ProofOfSpace
@@ -51,9 +51,11 @@ from chia.types.blockchain_format.proof_of_space import (
 
 class FarmerAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.farmer_stub import FarmerApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("FarmerAPI", None)
+        # Verify this class implements the FarmerApiStub protocol
+        def _protocol_check(self: FarmerAPI) -> FarmerApiStub:
+            return self
 
     log: logging.Logger
     farmer: Farmer

chia/farmer/start_farmer.py

@@ -7,7 +7,7 @@ from typing import Any, Optional
 
 from chia_rs import ConsensusConstants
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.farmer.farmer import Farmer
@@ -68,7 +68,7 @@ def create_farmer_service(
         network_id=network_id,
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/full_node/full_node_api.py

@@ -131,9 +131,11 @@ async def tx_request_and_timeout(full_node: FullNode, transaction_id: bytes32, t
 
 class FullNodeAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.full_node_stub import FullNodeApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("FullNodeAPI", None)
+        # Verify this class implements the FullNodeApiStub protocol
+        def _protocol_check(self: FullNodeAPI) -> FullNodeApiStub:
+            return self
 
     log: logging.Logger
     full_node: FullNode

chia/full_node/start_full_node.py

@@ -9,7 +9,7 @@ from typing import Any, Optional
 from chia_rs import ConsensusConstants
 from chia_rs.sized_ints import uint16
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.full_node.full_node import FullNode
@@ -71,7 +71,7 @@ async def create_full_node_service(
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
         override_capabilities=override_capabilities,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/harvester/harvester_api.py

@@ -5,7 +5,7 @@ import logging
 import time
 from collections.abc import Awaitable, Sequence
 from pathlib import Path
-from typing import TYPE_CHECKING, ClassVar, Optional, cast
+from typing import TYPE_CHECKING, ClassVar, Optional
 
 from chia_rs import AugSchemeMPL, G1Element, G2Element, ProofOfSpace
 from chia_rs.sized_bytes import bytes32
@@ -38,9 +38,11 @@ from chia.wallet.derive_keys import master_sk_to_local_sk
 
 class HarvesterAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.harvester_stub import HarvesterApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("HarvesterAPI", None)
+        # Verify this class implements the HarvesterApiStub protocol
+        def _protocol_check(self: HarvesterAPI) -> HarvesterApiStub:
+            return self
 
     log: logging.Logger
     harvester: Harvester

chia/harvester/start_harvester.py

@@ -7,7 +7,7 @@ from typing import Any, Optional
 
 from chia_rs import ConsensusConstants
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.harvester.harvester import Harvester
@@ -64,7 +64,7 @@ def create_harvester_service(
         network_id=network_id,
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/introducer/introducer_api.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, ClassVar, Optional, cast
+from typing import TYPE_CHECKING, ClassVar, Optional
 
 from chia_rs.sized_ints import uint64
 
@@ -17,9 +17,11 @@ from chia.types.peer_info import TimestampedPeerInfo
 
 class IntroducerAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.introducer_stub import IntroducerApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("IntroducerAPI", None)
+        # Verify this class implements the IntroducerApiStub protocol
+        def _protocol_check(self: IntroducerAPI) -> IntroducerApiStub:
+            return self
 
     log: logging.Logger
     introducer: Introducer

chia/introducer/start_introducer.py

@@ -5,7 +5,7 @@ import pathlib
 import sys
 from typing import Any, Optional
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.introducer.introducer import Introducer
 from chia.introducer.introducer_api import IntroducerAPI
 from chia.introducer.introducer_service import IntroducerService
@@ -60,7 +60,7 @@ def create_introducer_service(
         service_name=SERVICE_NAME,
         network_id=network_id,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/seeder/start_crawler.py

@@ -8,7 +8,7 @@ from typing import Any, Optional
 
 from chia_rs import ConsensusConstants
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS
 from chia.protocols.outbound_message import NodeType
@@ -63,7 +63,7 @@ def create_full_node_crawler_service(
         network_id=network_id,
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/server/server.py

@@ -31,7 +31,7 @@ from chia.protocols.outbound_message import Message, NodeType
 from chia.protocols.protocol_message_types import ProtocolMessageTypes
 from chia.protocols.protocol_state_machine import message_requires_reply
 from chia.protocols.protocol_timing import INVALID_PROTOCOL_BAN_SECONDS
-from chia.server.api_protocol import ApiProtocol
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
 from chia.server.introducer_peers import IntroducerPeers
 from chia.server.ssl_context import private_ssl_paths, public_ssl_paths
 from chia.server.ws_connection import ConnectionCallback, WSChiaConnection
@@ -120,7 +120,7 @@ class ChiaServer:
     ssl_client_context: ssl.SSLContext
     node_id: bytes32
     exempt_peer_networks: list[Union[IPv4Network, IPv6Network]]
-    class_for_type: dict[NodeType, type[ApiProtocol]]
+    stub_metadata_for_type: dict[NodeType, ApiMetadata]
     all_connections: dict[bytes32, WSChiaConnection] = field(default_factory=dict)
     on_connect: Optional[ConnectionCallback] = None
     shut_down_event: asyncio.Event = field(default_factory=asyncio.Event)
@@ -148,7 +148,7 @@ class ChiaServer:
         config: dict[str, Any],
         private_ca_crt_key: tuple[Path, Path],
         chia_ca_crt_key: tuple[Path, Path],
-        class_for_type: dict[NodeType, type[ApiProtocol]],
+        stub_metadata_for_type: dict[NodeType, ApiMetadata],
         name: str = __name__,
     ) -> ChiaServer:
         log = logging.getLogger(name)
@@ -229,7 +229,7 @@ class ChiaServer:
             node_id=calculate_node_id(node_id_cert_path),
             exempt_peer_networks=[ip_network(net, strict=False) for net in config.get("exempt_peer_networks", [])],
             introducer_peers=IntroducerPeers() if local_type is NodeType.INTRODUCER else None,
-            class_for_type=class_for_type,
+            stub_metadata_for_type=stub_metadata_for_type,
         )
 
     def set_received_message_callback(self, callback: ConnectionCallback) -> None:
@@ -329,7 +329,7 @@ class ChiaServer:
                 inbound_rate_limit_percent=self._inbound_rate_limit_percent,
                 outbound_rate_limit_percent=self._outbound_rate_limit_percent,
                 local_capabilities_for_handshake=self._local_capabilities_for_handshake,
-                class_for_type=self.class_for_type,
+                stub_metadata_for_type=self.stub_metadata_for_type,
             )
             await connection.perform_handshake(self._network_id, self.get_port(), self._local_type)
             assert connection.connection_type is not None, "handshake failed to set connection type, still None"
@@ -480,7 +480,7 @@ class ChiaServer:
                 inbound_rate_limit_percent=self._inbound_rate_limit_percent,
                 outbound_rate_limit_percent=self._outbound_rate_limit_percent,
                 local_capabilities_for_handshake=self._local_capabilities_for_handshake,
-                class_for_type=self.class_for_type,
+                stub_metadata_for_type=self.stub_metadata_for_type,
                 session=session,
             )
             await connection.perform_handshake(self._network_id, server_port, self._local_type)

chia/server/start_service.py

@@ -16,7 +16,7 @@ from chia.daemon.server import service_launch_lock_path
 from chia.protocols.outbound_message import NodeType
 from chia.protocols.shared_protocol import default_capabilities
 from chia.rpc.rpc_server import RpcApiProtocol, RpcServer, RpcServiceProtocol, start_rpc_server
-from chia.server.api_protocol import ApiProtocol
+from chia.server.api_protocol import ApiMetadata, ApiProtocol
 from chia.server.chia_policy import set_chia_policy
 from chia.server.server import ChiaServer
 from chia.server.signal_handlers import SignalHandlers
@@ -61,7 +61,7 @@ class Service(Generic[_T_RpcServiceProtocol, _T_ApiProtocol, _T_RpcApiProtocol])
         network_id: str,
         *,
         config: dict[str, Any],
-        class_for_type: dict[NodeType, type[ApiProtocol]],
+        stub_metadata_for_type: dict[NodeType, ApiMetadata],
         upnp_ports: Optional[list[int]] = None,
         connect_peers: Optional[set[UnresolvedPeerInfo]] = None,
         on_connect_callback: Optional[Callable[[WSChiaConnection], Awaitable[None]]] = None,
@@ -123,7 +123,7 @@ class Service(Generic[_T_RpcServiceProtocol, _T_ApiProtocol, _T_RpcApiProtocol])
             self.service_config,
             (private_ca_crt, private_ca_key),
             (chia_ca_crt, chia_ca_key),
-            class_for_type=class_for_type,
+            stub_metadata_for_type=stub_metadata_for_type,
             name=f"{service_name}_server",
         )
         f = getattr(node, "set_server", None)

chia/server/ws_connection.py

@@ -84,7 +84,7 @@ class WSChiaConnection:
     close_callback: Optional[ConnectionClosedCallbackProtocol] = field(repr=False)
     outbound_rate_limiter: RateLimiter
     inbound_rate_limiter: RateLimiter
-    class_for_type: dict[NodeType, type[ApiProtocol]] = field(repr=False)
+    stub_metadata_for_type: dict[NodeType, ApiMetadata] = field(repr=False)
 
     # connection properties
     is_outbound: bool
@@ -140,7 +140,7 @@ class WSChiaConnection:
         inbound_rate_limit_percent: int,
         outbound_rate_limit_percent: int,
         local_capabilities_for_handshake: list[tuple[uint16, str]],
-        class_for_type: dict[NodeType, type[ApiProtocol]],
+        stub_metadata_for_type: dict[NodeType, ApiMetadata],
         session: Optional[ClientSession] = None,
     ) -> WSChiaConnection:
         assert ws._writer is not None
@@ -172,7 +172,7 @@ class WSChiaConnection:
             inbound_rate_limiter=RateLimiter(incoming=True, percentage_of_limit=inbound_rate_limit_percent),
             is_outbound=is_outbound,
             received_message_callback=received_message_callback,
-            class_for_type=class_for_type,
+            stub_metadata_for_type=stub_metadata_for_type,
             session=session,
         )
 
@@ -407,7 +407,7 @@ class WSChiaConnection:
 
             if full_message.type == ProtocolMessageTypes.error.value:
                 error = Error.from_bytes(full_message.data)
-                self.api.log.warning(f"ApiError: {error} from {self.peer_node_id}, {self.peer_info}")
+                self.log.warning(f"ApiError: {error} from {self.peer_node_id}, {self.peer_info}")
                 return None
 
             bare_message_type = ProtocolMessageTypes(full_message.type)
@@ -545,7 +545,7 @@ class WSChiaConnection:
         assert request_metadata is not None, f"ApiMetadata unavailable for {request_method}"
         if (
             request_metadata.request_type
-            not in self.class_for_type[self.connection_type].metadata.message_type_to_request
+            not in self.stub_metadata_for_type[self.connection_type].message_type_to_request
         ):
             raise AttributeError(
                 f"Node type {self.connection_type} does not have method {request_metadata.request_type.name}"
@@ -572,7 +572,7 @@ class WSChiaConnection:
             await self.ban_peer_bad_protocol(error_message)
             raise ProtocolError(Err.INVALID_PROTOCOL_MESSAGE, [error_message])
 
-        recv_method = self.class_for_type[self.local_type].metadata.message_type_to_request[recv_message_type].method
+        recv_method = self.stub_metadata_for_type[self.local_type].message_type_to_request[recv_message_type].method
         receive_metadata = ApiMetadata.from_bound_method(recv_method)
         assert receive_metadata is not None, f"ApiMetadata unavailable for {recv_method}"
         return receive_metadata.message_class.from_bytes(response.data)

chia/simulator/start_simulator.py

@@ -10,7 +10,7 @@ from typing import Any, Optional
 from chia_rs.sized_bytes import bytes32
 from chia_rs.sized_ints import uint16
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.full_node.full_node import FullNode
 from chia.protocols.outbound_message import NodeType
 from chia.server.signal_handlers import SignalHandlers
@@ -65,7 +65,7 @@ async def create_full_node_simulator_service(
         rpc_info=(SimulatorFullNodeRpcApi, service_config["rpc_port"]),
         connect_to_daemon=connect_to_daemon,
         override_capabilities=override_capabilities,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/solver/solver_api.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, ClassVar, Optional, cast
+from typing import TYPE_CHECKING, ClassVar, Optional
 
 from chia.protocols.outbound_message import Message, make_msg
 from chia.protocols.protocol_message_types import ProtocolMessageTypes
@@ -12,9 +12,11 @@ from chia.solver.solver import Solver
 
 class SolverAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.solver_stub import SolverApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("SolverAPI", None)
+        # Verify this class implements the SolverApiStub protocol
+        def _protocol_check(self: SolverAPI) -> SolverApiStub:
+            return self
 
     log: logging.Logger
     solver: Solver

chia/solver/start_solver.py

@@ -9,7 +9,7 @@ from typing import Any, Optional
 from chia_rs import ConsensusConstants
 from chia_rs.sized_ints import uint16
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.protocols.outbound_message import NodeType
@@ -66,7 +66,7 @@ def create_solver_service(
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
         override_capabilities=override_capabilities,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/timelord/start_timelord.py

@@ -7,7 +7,7 @@ from typing import Any, Optional
 
 from chia_rs import ConsensusConstants
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.protocols.outbound_message import NodeType
@@ -61,7 +61,7 @@ def create_timelord_service(
         network_id=network_id,
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/timelord/timelord_api.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import logging
 import time
-from typing import TYPE_CHECKING, ClassVar, Optional, cast
+from typing import TYPE_CHECKING, ClassVar, Optional
 
 from chia_rs.sized_ints import uint64
 
@@ -19,9 +19,11 @@ log = logging.getLogger(__name__)
 
 class TimelordAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.timelord_stub import TimelordApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("TimelordAPI", None)
+        # Verify this class implements the TimelordApiStub protocol
+        def _protocol_check(self: TimelordAPI) -> TimelordApiStub:
+            return self
 
     log: logging.Logger
     timelord: Timelord

chia/wallet/start_wallet.py

@@ -8,7 +8,7 @@ from typing import Any, Optional
 
 from chia_rs import ConsensusConstants
 
-from chia.apis import ApiProtocolRegistry
+from chia.apis import StubMetadataRegistry
 from chia.consensus.constants import replace_str_to_bytes
 from chia.consensus.default_constants import DEFAULT_CONSTANTS, update_testnet_overrides
 from chia.protocols.outbound_message import NodeType
@@ -72,7 +72,7 @@ def create_wallet_service(
         network_id=network_id,
         rpc_info=rpc_info,
         connect_to_daemon=connect_to_daemon,
-        class_for_type=ApiProtocolRegistry,
+        stub_metadata_for_type=StubMetadataRegistry,
     )
 
 

chia/wallet/wallet_node_api.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, ClassVar, cast
+from typing import TYPE_CHECKING, ClassVar
 
 from chia_rs import RespondToPhUpdates
 
@@ -16,9 +16,11 @@ from chia.wallet.wallet_node import WalletNode
 
 class WalletNodeAPI:
     if TYPE_CHECKING:
-        from chia.server.api_protocol import ApiProtocol
+        from chia.apis.wallet_stub import WalletNodeApiStub
 
-        _protocol_check: ClassVar[ApiProtocol] = cast("WalletNodeAPI", None)
+        # Verify this class implements the WalletNodeApiStub protocol
+        def _protocol_check(self: WalletNodeAPI) -> WalletNodeApiStub:
+            return self
 
     log: logging.Logger
     wallet_node: WalletNode

tach.toml

@@ -52,6 +52,7 @@ depends_on = [
     { path = "chia.apis", deprecated = false },
     { path = "chia.cmds", deprecated = false },
     { path = "chia.protocols", deprecated = false },
+    { path = "chia.solver", deprecated = false },
     { path = "chia.timelord", deprecated = false },
     { path = "chia.harvester", deprecated = false },
     { path = "chia.daemon", deprecated = false },
@@ -88,12 +89,6 @@ depends_on = [
 [[modules]]
 path = "chia.apis"
 depends_on = [
-    { path = "chia.harvester", deprecated = false },
-    { path = "chia.farmer", deprecated = false },
-    { path = "chia.introducer", deprecated = false },
-    { path = "chia.wallet", deprecated = false },
-    { path = "chia.full_node", deprecated = false },
-    { path = "chia.timelord", deprecated = false },
     { path = "chia.protocols", deprecated = false },
     { path = "chia.server", deprecated = false },
 ]
@@ -210,6 +205,7 @@ depends_on = [
     { path = "chia.rpc", deprecated = false },
     { path = "chia.server", deprecated = false },
     { path = "chia.simulator", deprecated = false },
+    { path = "chia.solver", deprecated = false },
     { path = "chia.protocols", deprecated = false },
     { path = "chia.data_layer", deprecated = false },
 ]
@@ -263,6 +259,18 @@ depends_on = [
     { path = "chia.daemon", deprecated = false },
 ]
 
+[[modules]]
+path = "chia.solver"
+depends_on = [
+    "chia.protocols",
+    "chia.types",
+    "chia.util",
+    { path = "chia.apis", deprecated = false },
+    { path = "chia.consensus", deprecated = false },
+    { path = "chia.rpc", deprecated = false },
+    { path = "chia.server", deprecated = false },
+]
+
 [[modules]]
 path = "chia.seeder"
 depends_on = [