Types

pyvider.rpcplugin.types

Type definitions for the Pyvider RPC plugin system.

This module provides Protocol classes, TypeVars, and type aliases that define the interfaces and contracts used throughout the pyvider.rpcplugin package. These types enable static type checking and clear API boundaries.

For most users, these types are used only in type annotations. Advanced users implementing custom protocol handlers will need to implement the Protocol interfaces defined here.

Classes

ConnectionT

Bases: Protocol

Protocol for transport connections.

This protocol defines the minimal interface for connection objects used by transport implementations.

Functions

close async

close() -> None

Close the connection and clean up resources.

Source code in pyvider/rpcplugin/types.py

async def close(self) -> None:
    """
    Close the connection and clean up resources.
    """
    ...  # pragma: no cover

receive_data async

receive_data(size: int = 16384) -> bytes

Receive data from the connection.

Parameters:

Name	Type	Description	Default
size	int	Maximum number of bytes to receive	16384

Returns:

Type	Description
bytes	Received data as bytes

Source code in pyvider/rpcplugin/types.py

async def receive_data(self, size: int = 16384) -> bytes:
    """
    Receive data from the connection.

    Args:
        size: Maximum number of bytes to receive

    Returns:
        Received data as bytes
    """
    ...  # pragma: no cover

send_data async

send_data(data: bytes) -> None

Send data over the connection.

Parameters:

Name	Type	Description	Default
data	bytes	Bytes to send	required

Source code in pyvider/rpcplugin/types.py

async def send_data(self, data: bytes) -> None:
    """
    Send data over the connection.

    Args:
        data: Bytes to send
    """
    ...  # pragma: no cover

RPCPluginHandler

Bases: Protocol

Protocol defining the interface that all RPC handlers must implement.

This is a runtime-checkable protocol that defines the minimal interface required for a class to serve as a handler for an RPC plugin. The actual methods required will depend on the specific gRPC service being implemented.

RPCPluginProtocol

Bases: Protocol

Protocol defining the interface that all RPC protocol implementations must follow.

This protocol defines the contract for protocol implementations that bridge between gRPC services and Pyvider's RPC plugin system.

Functions

add_to_server async

add_to_server(server: Any, handler: Any) -> None

Adds the protocol implementation to the gRPC server.

Parameters:

Name	Type	Description	Default
server	Any	The gRPC async server instance	required
handler	Any	The handler implementing the RPC methods	required

Source code in pyvider/rpcplugin/types.py

async def add_to_server(self, server: Any, handler: Any) -> None:
    """
    Adds the protocol implementation to the gRPC server.

    Args:
        server: The gRPC async server instance
        handler: The handler implementing the RPC methods
    """
    ...  # pragma: no cover

get_grpc_descriptors async

get_grpc_descriptors() -> tuple[Any, str]

Returns the protobuf descriptor set and service name.

Returns:

Type	Description
tuple[Any, str]	Tuple containing the protobuf descriptor module and service name string.

Source code in pyvider/rpcplugin/types.py

async def get_grpc_descriptors(self) -> tuple[Any, str]:
    """
    Returns the protobuf descriptor set and service name.

    Returns:
        Tuple containing the protobuf descriptor module and service name string.
    """
    ...  # pragma: no cover

get_method_type

get_method_type(method_name: str) -> str

Gets the gRPC method type for a given method name.

Parameters:

Name	Type	Description	Default
method_name	str	The full method path (e.g., "/plugin.GRPCStdio/StreamStdio")	required

Returns:

Type	Description
str	String representing the method type (e.g., "unary_unary", "stream_stream")

Source code in pyvider/rpcplugin/types.py

def get_method_type(self, method_name: str) -> str:
    """
    Gets the gRPC method type for a given method name.

    Args:
        method_name: The full method path (e.g., "/plugin.GRPCStdio/StreamStdio")

    Returns:
        String representing the method type (e.g., "unary_unary", "stream_stream")
    """
    ...  # pragma: no cover

RPCPluginTransport

Bases: Protocol

Protocol defining the interface that all transport implementations must follow.

This protocol defines the contract for transport implementations that handle the low-level network communication between RPC plugin components.

Functions

close async

close() -> None

Close the transport and clean up resources.

Source code in pyvider/rpcplugin/types.py

async def close(self) -> None:
    """
    Close the transport and clean up resources.
    """
    ...  # pragma: no cover

connect async

connect(endpoint: str) -> None

Connect to a remote endpoint.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to connect to	required

Source code in pyvider/rpcplugin/types.py

async def connect(self, endpoint: str) -> None:
    """
    Connect to a remote endpoint.

    Args:
        endpoint: The endpoint to connect to
    """
    ...  # pragma: no cover

listen async

listen() -> str

Start listening for connections and return the endpoint.

Returns:

Type	Description
str	String representation of the endpoint (e.g., "unix:/tmp/socket" or
str	"127.0.0.1:50051")

Source code in pyvider/rpcplugin/types.py

async def listen(self) -> str:
    """
    Start listening for connections and return the endpoint.

    Returns:
        String representation of the endpoint (e.g., "unix:/tmp/socket" or
        "127.0.0.1:50051")
    """
    ...  # pragma: no cover

SecureRpcClientT

Bases: Protocol

Protocol for an RPC client supporting secure transport and handshake.

This protocol defines the interface for clients that support secure communication with mTLS and proper handshake negotiation.

Functions

close async

close() -> None

Close the client connection and clean up resources.

Source code in pyvider/rpcplugin/types.py

async def close(self) -> None:
    """Close the client connection and clean up resources."""
    ...  # pragma: no cover

SerializableT

Bases: Protocol

Protocol for objects that can be serialized to/from dict.

This protocol defines the minimal interface for objects that can be serialized to and from dictionary representations.

Functions

from_dict classmethod

from_dict(data: dict[str, Any]) -> SerializableT

Create an object from a dictionary representation.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary containing object data	required

Returns:

Type	Description
SerializableT	New instance of the class

Source code in pyvider/rpcplugin/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> SerializableT:
    """
    Create an object from a dictionary representation.

    Args:
        data: Dictionary containing object data

    Returns:
        New instance of the class
    """
    ...  # pragma: no cover

to_dict

to_dict() -> dict[str, Any]

Convert the object to a dictionary representation.

Returns:

Type	Description
dict[str, Any]	Dictionary representation of the object

Source code in pyvider/rpcplugin/types.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert the object to a dictionary representation.

    Returns:
        Dictionary representation of the object
    """
    ...  # pragma: no cover

Functions

is_valid_handler

is_valid_handler(obj: Any) -> TypeGuard[RPCPluginHandler]

TypeGuard that checks if an object implements the RPCPluginHandler protocol.

Parameters:

Name	Type	Description	Default
obj	Any	The object to check	required

Returns:

Type	Description
TypeGuard[RPCPluginHandler]	True if the object implements RPCPluginHandler, False otherwise

Source code in pyvider/rpcplugin/types.py

def is_valid_handler(obj: Any) -> TypeGuard[RPCPluginHandler]:
    """
    TypeGuard that checks if an object implements the RPCPluginHandler protocol.

    Args:
        obj: The object to check

    Returns:
        True if the object implements RPCPluginHandler, False otherwise
    """
    return isinstance(obj, RPCPluginHandler)

is_valid_protocol

is_valid_protocol(obj: Any) -> TypeGuard[RPCPluginProtocol]

TypeGuard that checks if an object implements the RPCPluginProtocol protocol.

Parameters:

Name	Type	Description	Default
obj	Any	The object to check	required

Returns:

Type	Description
TypeGuard[RPCPluginProtocol]	True if the object implements RPCPluginProtocol, False otherwise

Source code in pyvider/rpcplugin/types.py

def is_valid_protocol(obj: Any) -> TypeGuard[RPCPluginProtocol]:
    """
    TypeGuard that checks if an object implements the RPCPluginProtocol protocol.

    Args:
        obj: The object to check

    Returns:
        True if the object implements RPCPluginProtocol, False otherwise
    """
    return isinstance(obj, RPCPluginProtocol)

is_valid_transport

is_valid_transport(
    obj: Any,
) -> TypeGuard[RPCPluginTransport]

TypeGuard that checks if an object implements the RPCPluginTransport protocol.

Parameters:

Name	Type	Description	Default
obj	Any	The object to check	required

Returns:

Type	Description
TypeGuard[RPCPluginTransport]	True if the object implements RPCPluginTransport, False otherwise

Source code in pyvider/rpcplugin/types.py

def is_valid_transport(obj: Any) -> TypeGuard[RPCPluginTransport]:
    """
    TypeGuard that checks if an object implements the RPCPluginTransport protocol.

    Args:
        obj: The object to check

    Returns:
        True if the object implements RPCPluginTransport, False otherwise
    """
    return isinstance(obj, RPCPluginTransport)