Index

pyvider.rpcplugin

Pyvider RPC Plugin Package.

This package exports the main classes and exceptions for the Pyvider RPC Plugin system, making them available for direct import from pyvider.rpcplugin.

Classes

ConfigError

ConfigError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Configuration-related errors in the RPC plugin system.

Raised when there are issues with plugin configuration, including: - Invalid configuration values - Missing required configuration - Type mismatches in configuration - Environment variable parsing errors - Configuration validation failures

Example

from pyvider.rpcplugin.exception import ConfigError

# Raise when required config is missing
if not config.get("plugin_magic_cookie_value"):
    raise ConfigError(
        "Magic cookie value is required",
        hint="Set PLUGIN_MAGIC_COOKIE_VALUE environment variable",
        code="CONFIG_MISSING_COOKIE"
    )

# Raise when config value is invalid
if port < 0 or port > 65535:
    raise ConfigError(
        f"Invalid port number: {port}",
        hint="Port must be between 0 and 65535",
        code="CONFIG_INVALID_PORT"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

HandshakeError

HandshakeError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Handshake protocol errors during plugin initialization.

Raised when the handshake protocol fails, including: - Magic cookie validation failures - Protocol version negotiation failures - Transport negotiation failures - Certificate exchange errors - Timeout during handshake - Malformed handshake data

Example

from pyvider.rpcplugin.exception import HandshakeError

# Raise when magic cookie doesn't match
if received_cookie != expected_cookie:
    raise HandshakeError(
        "Magic cookie mismatch",
        hint="Ensure client and server use the same PLUGIN_MAGIC_COOKIE_VALUE",
        code="HANDSHAKE_COOKIE_MISMATCH"
    )

# Raise when no compatible protocol version
if not compatible_versions:
    raise HandshakeError(
        f"No compatible protocol versions: client={client_versions}, server={server_versions}",
        hint="Update client or server to support compatible protocol versions",
        code="HANDSHAKE_VERSION_MISMATCH"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

ProtocolError

ProtocolError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Protocol violation errors in RPC communication.

Raised when the plugin protocol is violated, including: - Invalid message format - Unexpected message types - Protocol state violations - Missing required protocol fields - Incompatible protocol operations

Example

from pyvider.rpcplugin.exception import ProtocolError

# Raise when message format is invalid
if not hasattr(message, 'required_field'):
    raise ProtocolError(
        "Protocol message missing required field",
        hint="Ensure message conforms to protocol specification",
        code="PROTOCOL_INVALID_MESSAGE"
    )

# Raise when protocol state is violated
if not self.handshake_complete:
    raise ProtocolError(
        "Cannot send data before handshake completion",
        hint="Wait for handshake to complete before sending messages",
        code="PROTOCOL_STATE_ERROR"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

RPCPluginClient

Bases: ClientHandshakeMixin, ClientProcessMixin

Client interface for interacting with Terraform-compatible plugin servers.

The RPCPluginClient handles the complete lifecycle of plugin communication: 1. Launching or attaching to a plugin server subprocess 2. Performing handshake, protocol negotiation, and transport selection 3. Setting up secure TLS/mTLS communication when enabled 4. Creating gRPC channels and service stubs 5. Providing plugin logs (stdout/stderr) streaming 6. Managing broker subchannels for multi-service communication 7. Handling graceful shutdown of plugin processes

The client follows the Terraform go-plugin protocol, which includes a standardized handshake format, negotiated protocol version, and support for Unix socket or TCP transport modes.

Attributes:

Name	Type	Description
command	list[str]	List containing the plugin executable command and arguments
config	dict[str, Any] | None	Optional configuration dictionary for customizing client behavior

Example

# Create a client for a plugin
client = RPCPluginClient(
    command=["terraform-provider-example"],
    config={"env": {"TF_LOG": "DEBUG"}}
)

# Start the client (launches process, performs handshake, etc.)
await client.start()

# Use the created channel with protocol-specific stubs
provider_stub = MyProviderStub(client.grpc_channel)
response = await provider_stub.SomeMethod(request)

# Graceful shutdown
await client.shutdown_plugin()
await client.close()

Note

The client supports automatic mTLS if enabled in configuration, and can read/generate certificates as needed for secure communication.

Functions

__aenter__ async

__aenter__() -> RPCPluginClient

Async context manager entry.

Source code in pyvider/rpcplugin/client/core.py

async def __aenter__(self) -> RPCPluginClient:
    """Async context manager entry."""
    await self.start()
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Async context manager exit with cleanup.

Source code in pyvider/rpcplugin/client/core.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context manager exit with cleanup."""
    try:
        await self.shutdown_plugin()
    except Exception as e:
        self.logger.warning(f"⚠️ Error during shutdown in context manager: {e}", exc_info=True)
    finally:
        await self.close()

__attrs_post_init__

__attrs_post_init__() -> None

Initialize client state after attributes are set.

Source code in pyvider/rpcplugin/client/core.py

def __attrs_post_init__(self) -> None:
    """
    Initialize client state after attributes are set.
    """
    self.logger = logger

close async

close() -> None

Close the client connection and clean up all resources.

This method performs a complete cleanup of the client state, including stopping tasks, closing channels, terminating processes, and cleaning up transport resources.

Source code in pyvider/rpcplugin/client/core.py

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

    This method performs a complete cleanup of the client state,
    including stopping tasks, closing channels, terminating processes,
    and cleaning up transport resources.
    """
    self.logger.debug("🔒 Closing RPCPluginClient...")

    await self._cancel_tasks()
    await self._close_grpc_channel()
    await self._terminate_process()
    await self._close_transport()
    self._reset_state()

shutdown_plugin async

shutdown_plugin() -> None

Gracefully shutdown the plugin server through gRPC controller.

This method sends a shutdown signal to the plugin server, allowing it to clean up resources before termination.

Source code in pyvider/rpcplugin/client/core.py

async def shutdown_plugin(self) -> None:
    """
    Gracefully shutdown the plugin server through gRPC controller.

    This method sends a shutdown signal to the plugin server, allowing it
    to clean up resources before termination.
    """
    try:
        if self._controller_stub:
            await self._controller_stub.Shutdown(ControllerEmpty())
            self.logger.debug("📤 Shutdown signal sent to plugin.")
        else:
            self.logger.warning("⚠️ No controller stub available for shutdown signal.")
    except grpc.RpcError:
        # Expected behavior when plugin shuts down immediately
        pass
    except Exception as e:
        self.logger.warning(f"⚠️ Error sending shutdown signal to plugin: {e}", exc_info=True)

    # Give the plugin a moment to shut down gracefully
    await asyncio.sleep(DEFAULT_CLEANUP_WAIT_TIME)

start async

start() -> None

Start the plugin client: launch process, perform handshake, create channel.

This is the main entry point for establishing communication with a plugin. It orchestrates the complete connection process.

Raises:

Type	Description
HandshakeError	If handshake fails
TransportError	If transport setup fails
ProtocolError	If protocol negotiation fails

Source code in pyvider/rpcplugin/client/core.py

async def start(self) -> None:
    """
    Start the plugin client: launch process, perform handshake, create channel.

    This is the main entry point for establishing communication with a plugin.
    It orchestrates the complete connection process.

    Raises:
        HandshakeError: If handshake fails
        TransportError: If transport setup fails
        ProtocolError: If protocol negotiation fails
    """
    self.logger.debug("🚀 Starting RPCPluginClient...")

    try:
        await self._connect_and_handshake_with_retry()
        self.is_started = True
    except Exception as e:
        self.logger.error(f"❌ Failed to start RPCPluginClient: {e}")
        self._handshake_failed_event.set()
        # Clean up any partial state on start failure
        await self.close()
        raise

RPCPluginConfig

Bases: RuntimeConfig

Comprehensive configuration for the RPC plugin system.

This configuration class extends Foundation's RuntimeConfig to provide type-safe, environment-aware configuration for all aspects of the RPC plugin framework. All fields support environment variable overrides with automatic type conversion and validation.

The configuration is organized into functional areas:

Core Settings: - Protocol version negotiation - Magic cookie authentication - Logging configuration

Transport Settings: - Connection and handshake timeouts - Buffer sizes for network I/O - Supported transport mechanisms (Unix, TCP)

Security Settings: - mTLS configuration and auto-generation - Certificate paths and validity periods - Insecure mode for development

gRPC Settings: - Keepalive parameters - Message size limits - Grace periods for shutdown

Client Settings: - Retry logic and backoff strategies - Maximum retry attempts - Transport preferences

Server Settings: - Host and port binding - Unix socket paths - Transport availability

Feature Settings: - Rate limiting with token bucket - Health check service - UI components

Example

from pyvider.rpcplugin.config import rpcplugin_config

# Access configuration values
if rpcplugin_config.plugin_auto_mtls:
    print("mTLS is enabled")

# Modify configuration
rpcplugin_config.plugin_log_level = "DEBUG"
rpcplugin_config.plugin_rate_limit_enabled = True

Note

All configuration fields can be overridden via environment variables. The env_field decorator ensures proper parsing and validation. Default values are defined in the defaults module for consistency.

RPCPluginError

RPCPluginError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: FoundationError

Base exception for all Pyvider RPC plugin errors.

This class serves as the root of the exception hierarchy for the plugin system. It can be subclassed to create more specific error types.

Attributes:

Name	Type	Description
message		A human-readable error message.
hint		An optional hint for resolving the error.
code		An optional error code associated with the error.

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

Functions

__str__

__str__() -> str

Format error message with prefix, code, and hint for backward compatibility.

Source code in pyvider/rpcplugin/exception.py

def __str__(self) -> str:
    """Format error message with prefix, code, and hint for backward compatibility."""
    prefix = f"[{self.__class__.__name__}]"

    # Get the base message from parent (which is just the message we passed)
    base_message = self.message

    # Ensure message is prefixed only if it's not already
    effective_message = base_message
    if not base_message.startswith("[") or not base_message.lower().startswith(prefix.lower()):
        effective_message = f"{prefix} {base_message}"

    parts = [effective_message]
    # Only add code if it was explicitly provided by the user
    if self.code is not None:
        parts.append(f"[Code: {self.code}]")
    if self.hint:
        parts.append(f"(Hint: {self.hint})")

    return " ".join(parts)

RPCPluginProtocol

Bases: ABC, Generic[ServerT, HandlerT]

Abstract base class for defining RPC protocols. ServerT: Type of gRPC server HandlerT: Type of handler implementation

Functions

add_to_server abstractmethod async

add_to_server(server: ServerT, handler: HandlerT) -> None

Adds the protocol implementation to the gRPC server.

Parameters:

Name	Type	Description	Default
server	ServerT	The gRPC async server instance.	required
handler	HandlerT	The handler implementing the RPC methods for this protocol.	required

Source code in pyvider/rpcplugin/protocol/base.py

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

    Args:
        server: The gRPC async server instance.
        handler: The handler implementing the RPC methods for this protocol.
    """
    pass

get_grpc_descriptors abstractmethod async

get_grpc_descriptors() -> tuple[Any, str]

Returns the protobuf descriptor set and service name.

Source code in pyvider/rpcplugin/protocol/base.py

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

RPCPluginServer

Bases: Generic[ServerT, HandlerT, TransportT], ServerNetworkMixin

Server interface for hosting Terraform-compatible plugin services.

The RPCPluginServer handles the complete lifecycle of plugin hosting: 1. Transport setup (Unix socket or TCP) with optional mTLS 2. Handshake protocol negotiation with clients 3. gRPC server initialization and service registration 4. Rate limiting and health check services 5. Signal handling for graceful shutdown 6. Optional shutdown file monitoring

The server follows the Terraform go-plugin protocol, which includes a standardized handshake format, negotiated protocol version, and support for Unix socket or TCP transport modes.

Attributes:

Name	Type	Description
protocol	RPCPluginProtocol[ServerT, HandlerT]	Protocol implementation for the plugin service
handler	HandlerT	Service handler instance for the protocol
config	dict[str, Any] | None	Optional configuration dictionary for customizing server behavior
transport	TransportT | None	Optional pre-configured transport instance

Example

# Create a server for a plugin
server = RPCPluginServer(
    protocol=MyProtocol(),
    handler=MyServiceHandler(),
    config={"PLUGIN_AUTO_MTLS": True}
)

# Start the server (setup transport, handshake, serve)
await server.serve()

Note

The server supports automatic mTLS if enabled in configuration, and can generate certificates as needed for secure communication.

Functions

serve async

serve() -> None

Start the plugin server and serve until shutdown.

This is the main entry point for running the server. It orchestrates the complete server lifecycle including transport setup, handshake, gRPC server initialization, and serving until shutdown.

Raises:

Type	Description
TransportError	If transport setup fails
ProtocolError	If handshake or protocol setup fails

Source code in pyvider/rpcplugin/server/core.py

async def serve(self) -> None:
    """
    Start the plugin server and serve until shutdown.

    This is the main entry point for running the server. It orchestrates
    the complete server lifecycle including transport setup, handshake,
    gRPC server initialization, and serving until shutdown.

    Raises:
        TransportError: If transport setup fails
        ProtocolError: If handshake or protocol setup fails
    """
    if _tracer:
        with _tracer.start_as_current_span("rpc.server.serve") as span:
            span.set_attribute("component", "server")
            await self._serve_impl()
    else:
        await self._serve_impl()

stop async

stop() -> None

Stop the server and clean up resources.

This method performs graceful shutdown of the server including stopping the gRPC server, cleaning up transport resources, and canceling background tasks.

Source code in pyvider/rpcplugin/server/core.py

async def stop(self) -> None:
    """
    Stop the server and clean up resources.

    This method performs graceful shutdown of the server including
    stopping the gRPC server, cleaning up transport resources,
    and canceling background tasks.
    """
    logger.info("🔒 Stopping RPCPluginServer...")

    # Cancel shutdown watcher task
    if self._shutdown_watcher_task and not self._shutdown_watcher_task.done():
        self._shutdown_watcher_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._shutdown_watcher_task

    # Stop gRPC server
    if self._server is not None:
        logger.debug("Stopping gRPC server...")
        server_to_stop = cast(grpc.aio.Server, self._server)
        await server_to_stop.stop(grace=0.5)
        self._server = None

    # Clean up transport
    if self._transport is not None:
        logger.debug("Closing transport...")
        transport_to_close = cast(RPCPluginTransportType, self._transport)
        await transport_to_close.close()
        self._transport = None

    # Complete the serving future if not already done
    if not self._serving_future.done():
        self._serving_future.set_result(None)

    # Exit if configured to do so (only in non-test environments)
    if self._exit_on_stop and not os.environ.get("PYTEST_CURRENT_TEST"):
        logger.info("⚡ Exiting process...")
        sys.exit(0)

wait_for_server_ready async

wait_for_server_ready(timeout: float | None = None) -> None

Wait for the server to be ready to accept connections.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum time to wait for server readiness	None

Raises:

Type	Description
TimeoutError	If server doesn't become ready within timeout
TransportError	If server setup fails

Source code in pyvider/rpcplugin/server/core.py

async def wait_for_server_ready(self, timeout: float | None = None) -> None:
    """
    Wait for the server to be ready to accept connections.

    Args:
        timeout: Maximum time to wait for server readiness

    Raises:
        TimeoutError: If server doesn't become ready within timeout
        TransportError: If server setup fails
    """
    if timeout is None:
        timeout = rpcplugin_config.plugin_server_ready_timeout

    logger.debug(f"Waiting for server to be ready (timeout: {timeout}s)")

    try:
        await asyncio.wait_for(self._serving_event.wait(), timeout=timeout)
        logger.debug("Server serving event is set")

        # Perform transport-specific readiness checks
        await self._verify_transport_readiness()

        logger.debug("Server is ready")
    except TimeoutError as e:
        error_msg = f"Server failed to become ready within {timeout} seconds"
        logger.error(error_msg)
        raise TimeoutError(error_msg) from e

SecurityError

SecurityError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Security-related errors in the plugin system.

Raised when security issues are detected, including: - Certificate validation failures - TLS/mTLS errors - Authentication failures - Authorization violations - Insecure configuration attempts - Certificate generation failures

Example

from pyvider.rpcplugin.exception import SecurityError

# Raise when certificate validation fails
if not validate_certificate(cert):
    raise SecurityError(
        "Certificate validation failed",
        hint="Check certificate expiration and CA trust chain",
        code="SECURITY_CERT_INVALID"
    )

# Raise when insecure operation is attempted
if config.plugin_insecure and config.plugin_auto_mtls:
    raise SecurityError(
        "Cannot enable both insecure mode and mTLS",
        hint="Choose either secure (mTLS) or insecure mode, not both",
        code="SECURITY_CONFIG_CONFLICT"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

TransportError

TransportError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Transport layer errors in plugin communication.

Raised when transport-level issues occur, including: - Connection failures - Socket errors - Network timeouts - Invalid endpoints - Transport initialization failures - I/O errors during communication

Example

from pyvider.rpcplugin.exception import TransportError

# Raise when connection fails
try:
    await transport.connect(endpoint)
except ConnectionRefusedError as e:
    raise TransportError(
        f"Failed to connect to {endpoint}",
        hint="Ensure the server is running and the endpoint is correct",
        code="TRANSPORT_CONNECTION_REFUSED"
    ) from e

# Raise when endpoint format is invalid
if not is_valid_endpoint(endpoint):
    raise TransportError(
        f"Invalid endpoint format: {endpoint}",
        hint="Use 'host:port' for TCP or 'unix:/path' for Unix sockets",
        code="TRANSPORT_INVALID_ENDPOINT"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

Functions

__getattr__

__getattr__(name: str) -> str

Support lazy loading of version.

This reduces initial import overhead by deferring version loading until first access.

Parameters:

Name	Type	Description	Default
name	str	Attribute name to lazy-load	required

Returns:

Type	Description
str	The attribute value

Raises:

Type	Description
AttributeError	If the attribute is not found

Source code in pyvider/rpcplugin/__init__.py

def __getattr__(name: str) -> str:
    """Support lazy loading of __version__.

    This reduces initial import overhead by deferring version loading
    until first access.

    Args:
        name: Attribute name to lazy-load

    Returns:
        The attribute value

    Raises:
        AttributeError: If the attribute is not found
    """
    if name == "__version__":
        from provide.foundation.utils.versioning import get_version

        return get_version("pyvider-rpcplugin", caller_file=__file__)
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

create_basic_protocol

create_basic_protocol() -> (
    type[RPCPluginProtocol[Any, Any]]
)

Create a basic RPCPluginProtocol class for testing or simple use cases.

This factory creates a minimal protocol implementation that can be used when no specific gRPC services are needed. It's useful for: - Testing the plugin framework - Creating placeholder protocols - Demonstrating the protocol interface

Returns:

Type	Description
type[RPCPluginProtocol[Any, Any]]	A class that implements RPCPluginProtocol with minimal functionality.
type[RPCPluginProtocol[Any, Any]]	The class can be instantiated with an optional service_name_override.

Example

# Create the protocol class
BasicProtocol = create_basic_protocol()

# Instantiate with custom service name
protocol = BasicProtocol(service_name_override="my.custom.service")

# Use with a server
server = plugin_server(protocol=protocol, handler=handler)

Note

The returned protocol class doesn't register any actual gRPC services. For production use, implement a custom protocol with real services.

Source code in pyvider/rpcplugin/factories.py

def create_basic_protocol() -> type[RPCPluginProtocol[Any, Any]]:
    """
    Create a basic RPCPluginProtocol class for testing or simple use cases.

    This factory creates a minimal protocol implementation that can be used
    when no specific gRPC services are needed. It's useful for:
    - Testing the plugin framework
    - Creating placeholder protocols
    - Demonstrating the protocol interface

    Returns:
        A class that implements RPCPluginProtocol with minimal functionality.
        The class can be instantiated with an optional service_name_override.

    Example:
        ```python
        # Create the protocol class
        BasicProtocol = create_basic_protocol()

        # Instantiate with custom service name
        protocol = BasicProtocol(service_name_override="my.custom.service")

        # Use with a server
        server = plugin_server(protocol=protocol, handler=handler)
        ```

    Note:
        The returned protocol class doesn't register any actual gRPC services.
        For production use, implement a custom protocol with real services.
    """

    class BasicRPCPluginProtocol(RPCPluginProtocol[Any, Any]):
        """
        Basic protocol implementation for testing and simple use cases.

        This protocol provides the minimal interface required by the framework
        but doesn't register any actual gRPC services. It's primarily used
        for testing or as a placeholder when no specific protocol is needed.

        Attributes:
            service_name: The name of the service (default: "pyvider.BasicRPCPluginProtocol")
        """

        service_name: str = "pyvider.BasicRPCPluginProtocol"

        def __init__(self, service_name_override: str | None = None) -> None:
            super().__init__()
            if service_name_override:
                self.service_name = service_name_override

        async def get_grpc_descriptors(self) -> tuple[Any, str]:
            logger.debug(f"BasicRPCPluginProtocol: get_grpc_descriptors for {self.service_name}")
            return (None, self.service_name)

        async def add_to_server(self, server: Any, handler: Any) -> None:
            logger.debug(
                f"BasicRPCPluginProtocol: add_to_server for {self.service_name} "
                "(no specific services added by this basic protocol itself)."
            )
            pass

        def get_method_type(self, method_name: str) -> str:
            logger.warning(
                f"BasicRPCPluginProtocol: get_method_type for {method_name} "
                "defaulting to unary_unary. Implement for specific protocols."
            )
            return "unary_unary"

    return BasicRPCPluginProtocol

plugin_client

plugin_client(
    command: list[str], config: dict[str, Any] | None = None
) -> RPCPluginClient

Factory for creating an RPC plugin client instance.

This factory creates a client that can launch and communicate with a plugin subprocess. The client handles the complete lifecycle of the plugin process, including launching, handshake, and cleanup.

Parameters:

Name	Type	Description	Default
command	list[str]	Command and arguments to launch the plugin process. Example: ["python", "my_plugin.py"] or ["./my-plugin"]	required
config	dict[str, Any] | None	Optional configuration dictionary. Currently only supports passing environment variables to the plugin subprocess via the "env" key: config={"env": {"VAR": "value"}}. To configure client behavior, set environment variables in your own process before creating the client.	None

Returns:

Type	Description
RPCPluginClient	An RPCPluginClient instance. You must call the async start()
RPCPluginClient	method to launch the plugin and establish connection.

Example

import asyncio
from pyvider.rpcplugin import plugin_client

async def main():
    # Create client for a Python plugin (recommended: use context manager)
    async with plugin_client(
        command=["python", "path/to/plugin.py"],
        config={"env": {"PLUGIN_LOG_LEVEL": "DEBUG"}}  # Pass env vars to plugin
    ) as client:
        # Launch plugin and establish connection
        await client.start()

        # Use the client's grpc_channel for RPC calls
        stub = MyServiceStub(client.grpc_channel)
        response = await stub.ProcessRequest(request)

        # Gracefully shutdown
        await client.shutdown_plugin()

    # Client automatically closed on context exit

asyncio.run(main())

Advanced Example

# Configure plugin subprocess environment variables
# Note: config dict is used to pass environment variables to the plugin process
async with plugin_client(
    command=["./my-secure-plugin"],
    config={
        "env": {
            "PLUGIN_LOG_LEVEL": "DEBUG",
            "PLUGIN_AUTO_MTLS": "true",
            "MY_PLUGIN_API_KEY": "secret-key",
        }
    }
) as client:
    await client.start()
    # Use client...
# Automatically closed on context exit

To configure the CLIENT behavior (not the plugin subprocess),

set environment variables in your own process:

import os

# Configure client retry behavior
os.environ["PLUGIN_CLIENT_MAX_RETRIES"] = "5"
os.environ["PLUGIN_CLIENT_RETRY_ENABLED"] = "true"
os.environ["PLUGIN_CONNECTION_TIMEOUT"] = "30.0"

# Configure client-side mTLS
os.environ["PLUGIN_CLIENT_CERT"] = "file:///path/to/client.crt"
os.environ["PLUGIN_CLIENT_KEY"] = "file:///path/to/client.key"

async with plugin_client(command=["./my-secure-plugin"]) as client:
    await client.start()
    # Use client...

Manual Cleanup (Alternative):

# For cases where you need manual control over lifecycle
client = plugin_client(command=["python", "plugin.py"])
try:
    await client.start()
    # Use client...
finally:
    await client.close()

Note

The client supports automatic retry with exponential backoff, subprocess management, stdio/stderr capture, and both Unix socket and TCP transports. The handshake protocol ensures secure communication via magic cookie validation.

Source code in pyvider/rpcplugin/factories.py

def plugin_client(
    command: list[str],
    config: dict[str, Any] | None = None,
) -> RPCPluginClient:
    """
    Factory for creating an RPC plugin client instance.

    This factory creates a client that can launch and communicate with
    a plugin subprocess. The client handles the complete lifecycle of
    the plugin process, including launching, handshake, and cleanup.

    Args:
        command: Command and arguments to launch the plugin process.
                Example: ["python", "my_plugin.py"] or ["./my-plugin"]
        config: Optional configuration dictionary. Currently only supports
               passing environment variables to the plugin subprocess via
               the "env" key: config={"env": {"VAR": "value"}}.
               To configure client behavior, set environment variables
               in your own process before creating the client.

    Returns:
        An RPCPluginClient instance. You must call the async start()
        method to launch the plugin and establish connection.

    Example:
        ```python
        import asyncio
        from pyvider.rpcplugin import plugin_client

        async def main():
            # Create client for a Python plugin (recommended: use context manager)
            async with plugin_client(
                command=["python", "path/to/plugin.py"],
                config={"env": {"PLUGIN_LOG_LEVEL": "DEBUG"}}  # Pass env vars to plugin
            ) as client:
                # Launch plugin and establish connection
                await client.start()

                # Use the client's grpc_channel for RPC calls
                stub = MyServiceStub(client.grpc_channel)
                response = await stub.ProcessRequest(request)

                # Gracefully shutdown
                await client.shutdown_plugin()

            # Client automatically closed on context exit

        asyncio.run(main())
        ```

    Advanced Example:
        ```python
        # Configure plugin subprocess environment variables
        # Note: config dict is used to pass environment variables to the plugin process
        async with plugin_client(
            command=["./my-secure-plugin"],
            config={
                "env": {
                    "PLUGIN_LOG_LEVEL": "DEBUG",
                    "PLUGIN_AUTO_MTLS": "true",
                    "MY_PLUGIN_API_KEY": "secret-key",
                }
            }
        ) as client:
            await client.start()
            # Use client...
        # Automatically closed on context exit
        ```

        # To configure the CLIENT behavior (not the plugin subprocess),
        # set environment variables in your own process:
        ```python
        import os

        # Configure client retry behavior
        os.environ["PLUGIN_CLIENT_MAX_RETRIES"] = "5"
        os.environ["PLUGIN_CLIENT_RETRY_ENABLED"] = "true"
        os.environ["PLUGIN_CONNECTION_TIMEOUT"] = "30.0"

        # Configure client-side mTLS
        os.environ["PLUGIN_CLIENT_CERT"] = "file:///path/to/client.crt"
        os.environ["PLUGIN_CLIENT_KEY"] = "file:///path/to/client.key"

        async with plugin_client(command=["./my-secure-plugin"]) as client:
            await client.start()
            # Use client...
        ```

    Manual Cleanup (Alternative):
        ```python
        # For cases where you need manual control over lifecycle
        client = plugin_client(command=["python", "plugin.py"])
        try:
            await client.start()
            # Use client...
        finally:
            await client.close()
        ```

    Note:
        The client supports automatic retry with exponential backoff,
        subprocess management, stdio/stderr capture, and both Unix
        socket and TCP transports. The handshake protocol ensures
        secure communication via magic cookie validation.
    """
    logger.debug(f"🏭 Creating plugin client for command: {command}")
    return RPCPluginClient(command=command, config=config or {})

plugin_protocol

plugin_protocol(
    protocol_class: type[PT_co] | None = None,
    handler_class: type[RPCPluginHandler] | None = None,
    service_name: str | None = None,
    **kwargs: Any
) -> PT_co

Factory for creating an RPC plugin protocol instance.

This factory provides a convenient way to instantiate protocol objects with proper configuration. It can either create a custom protocol instance or fall back to a basic protocol for testing.

Parameters:

Name	Type	Description	Default
protocol_class	type[PT_co] | None	Optional custom protocol class to instantiate. If None, creates a BasicRPCPluginProtocol.	None
handler_class	type[RPCPluginHandler] | None	Optional handler class (currently unused but reserved for future handler validation).	None
service_name	str | None	Optional service name to override the default. Passed as 'service_name_override' to the protocol.	None
**kwargs	Any	Additional keyword arguments passed to the protocol constructor.	{}

Returns:

Type	Description
PT_co	An instance of the specified protocol class, or BasicRPCPluginProtocol
PT_co	if no protocol_class was provided.

Example

# Create a basic protocol with default settings
protocol = plugin_protocol()

# Create a basic protocol with custom service name
protocol = plugin_protocol(service_name="my.custom.service")

# Create an instance of a custom protocol class
from my_plugin import MyCustomProtocol
protocol = plugin_protocol(
    protocol_class=MyCustomProtocol,
    service_name="my.service.v1",
    custom_option=True
)

# Use the protocol with a server
server = plugin_server(protocol=protocol, handler=handler)

Note

When using a custom protocol_class, ensure it accepts 'service_name_override' in its constructor if you want to use the service_name parameter.

Source code in pyvider/rpcplugin/factories.py

def plugin_protocol(
    protocol_class: type[PT_co] | None = None,  # PT_co bound to RPCPluginProtocol implicitly by usage
    handler_class: type[RPCPluginHandler]  # Use imported RPCPluginHandler
    | None = None,
    service_name: str | None = None,
    **kwargs: Any,  # Add **kwargs to accept arbitrary keyword arguments
) -> PT_co:
    """
    Factory for creating an RPC plugin protocol instance.

    This factory provides a convenient way to instantiate protocol objects
    with proper configuration. It can either create a custom protocol instance
    or fall back to a basic protocol for testing.

    Args:
        protocol_class: Optional custom protocol class to instantiate.
                       If None, creates a BasicRPCPluginProtocol.
        handler_class: Optional handler class (currently unused but reserved
                      for future handler validation).
        service_name: Optional service name to override the default.
                     Passed as 'service_name_override' to the protocol.
        **kwargs: Additional keyword arguments passed to the protocol constructor.

    Returns:
        An instance of the specified protocol class, or BasicRPCPluginProtocol
        if no protocol_class was provided.

    Example:
        ```python
        # Create a basic protocol with default settings
        protocol = plugin_protocol()

        # Create a basic protocol with custom service name
        protocol = plugin_protocol(service_name="my.custom.service")

        # Create an instance of a custom protocol class
        from my_plugin import MyCustomProtocol
        protocol = plugin_protocol(
            protocol_class=MyCustomProtocol,
            service_name="my.service.v1",
            custom_option=True
        )

        # Use the protocol with a server
        server = plugin_server(protocol=protocol, handler=handler)
        ```

    Note:
        When using a custom protocol_class, ensure it accepts 'service_name_override'
        in its constructor if you want to use the service_name parameter.
    """
    effective_protocol_class: type[PT_co]
    instance_kwargs = kwargs  # Initialize with all extra kwargs

    if protocol_class:
        effective_protocol_class = protocol_class
        # If service_name is provided, pass it as 'service_name_override'.
        # Custom protocols should handle 'service_name_override' for this factory
        # to configure their service name.
        if service_name:
            instance_kwargs["service_name_override"] = service_name
    else:
        # Default to BasicRPCPluginProtocol
        BasicProtoCls = create_basic_protocol()
        effective_protocol_class = cast(type[PT_co], BasicProtoCls)

        # For BasicRPCPluginProtocol, only 'service_name_override' is relevant.
        # Filter instance_kwargs to only pass this if service_name was provided,
        # or if 'service_name_override' was already in **kwargs from the call.
        final_basic_kwargs = {}
        if service_name:
            final_basic_kwargs["service_name_override"] = service_name
        elif "service_name_override" in instance_kwargs:
            # If service_name wasn't given directly to factory,
            # but was in **kwargs
            final_basic_kwargs["service_name_override"] = instance_kwargs["service_name_override"]
        instance_kwargs = final_basic_kwargs

    return effective_protocol_class(**instance_kwargs)

plugin_server

plugin_server(
    protocol: ProtocolT,
    handler: HandlerT,
    transport: str = "unix",
    transport_path: str | None = None,
    host: str = "127.0.0.1",
    port: int = 0,
    config: dict[str, Any] | None = None,
) -> RPCPluginServer[_ServerT, ServerHandlerT, _TransportT]

Factory for creating an RPC plugin server instance.

This factory simplifies server creation by handling transport setup and configuration. It supports both Unix socket and TCP transports with sensible defaults for each platform.

Parameters:

Name	Type	Description	Default
protocol	ProtocolT	The protocol instance defining the RPC services. Usually created with plugin_protocol().	required
handler	HandlerT	The service handler implementing the protocol's methods. This object will handle incoming RPC requests.	required
transport	str	Transport type to use. Either "unix" (default) or "tcp". Unix sockets are preferred for local IPC on Linux/macOS.	'unix'
transport_path	str | None	For Unix sockets, the socket file path. If None, a temporary path is generated.	None
host	str	For TCP transport, the bind address (default: "127.0.0.1").	'127.0.0.1'
port	int	For TCP transport, the port number (default: 0 for random).	0
config	dict[str, Any] | None	Optional configuration dictionary to override defaults. Can include settings like timeouts, buffer sizes, etc.	None

Returns:

Type	Description
RPCPluginServer[_ServerT, _HandlerT, _TransportT]	A configured RPCPluginServer instance ready to serve requests.
RPCPluginServer[_ServerT, _HandlerT, _TransportT]	Call the serve() method to start accepting connections.

Raises:

Type	Description
ValueError	If an unsupported transport type is specified.

Example

import asyncio
from pyvider.rpcplugin import plugin_server, plugin_protocol

class MyHandler:
    async def process(self, request):
        return {"result": "processed"}

async def main():
    # Create server with Unix socket (default)
    server = plugin_server(
        protocol=plugin_protocol(),
        handler=MyHandler()
    )

    # Or create TCP server
    server = plugin_server(
        protocol=plugin_protocol(),
        handler=MyHandler(),
        transport="tcp",
        port=8080
    )

    # Start serving
    await server.serve()

asyncio.run(main())

Note

The server will automatically handle the handshake protocol, including magic cookie validation and transport negotiation. For production use, consider enabling mTLS via configuration.

Source code in pyvider/rpcplugin/factories.py

def plugin_server(
    protocol: BaseProtocolTDefinition,
    handler: HandlerT,
    transport: str = "unix",
    transport_path: str | None = None,
    host: str = "127.0.0.1",
    port: int = 0,
    config: dict[str, Any] | None = None,
) -> RPCPluginServer[_ServerT, ServerHandlerT, _TransportT]:
    """
    Factory for creating an RPC plugin server instance.

    This factory simplifies server creation by handling transport setup
    and configuration. It supports both Unix socket and TCP transports
    with sensible defaults for each platform.

    Args:
        protocol: The protocol instance defining the RPC services.
                 Usually created with plugin_protocol().
        handler: The service handler implementing the protocol's methods.
                This object will handle incoming RPC requests.
        transport: Transport type to use. Either "unix" (default) or "tcp".
                  Unix sockets are preferred for local IPC on Linux/macOS.
        transport_path: For Unix sockets, the socket file path.
                       If None, a temporary path is generated.
        host: For TCP transport, the bind address (default: "127.0.0.1").
        port: For TCP transport, the port number (default: 0 for random).
        config: Optional configuration dictionary to override defaults.
               Can include settings like timeouts, buffer sizes, etc.

    Returns:
        A configured RPCPluginServer instance ready to serve requests.
        Call the serve() method to start accepting connections.

    Raises:
        ValueError: If an unsupported transport type is specified.

    Example:
        ```python
        import asyncio
        from pyvider.rpcplugin import plugin_server, plugin_protocol

        class MyHandler:
            async def process(self, request):
                return {"result": "processed"}

        async def main():
            # Create server with Unix socket (default)
            server = plugin_server(
                protocol=plugin_protocol(),
                handler=MyHandler()
            )

            # Or create TCP server
            server = plugin_server(
                protocol=plugin_protocol(),
                handler=MyHandler(),
                transport="tcp",
                port=8080
            )

            # Start serving
            await server.serve()

        asyncio.run(main())
        ```

    Note:
        The server will automatically handle the handshake protocol,
        including magic cookie validation and transport negotiation.
        For production use, consider enabling mTLS via configuration.
    """
    logger.debug(
        f"🏭 Creating plugin server: transport={transport}, path={transport_path}, host={host}, port={port}"
    )
    transport_instance: RPCPluginTransportType
    if transport == "unix":
        transport_instance = UnixSocketTransport(path=transport_path)
    elif transport == "tcp":
        transport_instance = TCPSocketTransport(host=host, port=port)
    else:
        raise ValueError(f"Unsupported transport type: {transport}")

    return RPCPluginServer(
        protocol=cast(BaseRpcAbcProtocol[_ServerT, ServerHandlerT], protocol),
        handler=cast(ServerHandlerT, handler),
        transport=cast(_TransportT, transport_instance),  # Use 'transport' kwarg
        config=config or {},
    )