SLIMRPC Compiler

The Slim RPC Compiler (protoc-slimrpc-plugin) is a protoc plugin that generates Python client stubs and server servicers for SLIMRPC (Slim RPC) from Protocol Buffer service definitions. This plugin enables you to build high-performance RPC services using the SLIMRPC framework.

Features

The Slim RPC Compiler has the following features:

• Generates Python client stubs for calling SlimRPC services
• Generates Python server servicers for implementing SlimRPC services
• Supports all gRPC streaming patterns: unary-unary, unary-stream, stream-unary, and stream-stream
• Compatible with both protoc and buf build systems
• Automatic import resolution for Protocol Buffer dependencies

Installation

There are two ways to install the Slim RPC Compiler:

Install via Cargo

cargo install agntcy-protoc-slimrpc-plugin

This will install the protoc-slimrpc-plugin binary to your Cargo bin directory (usually ~/.cargo/bin).

Compile from Source

1. Clone the repository:

   git clone https://github.com/agntcy/slim.git
   cd slim/data-plane/slimrpc-compiler
   

2. Build the plugin:

   cargo build --release
   

3. The compiled binary is available at data-plane/target/release/protoc-slimrpc-plugin.

Usage

Example Protocol Buffer Definition

Create a file called example.proto:

syntax = "proto3";

package example_service;

service Test {
  rpc ExampleUnaryUnary(ExampleRequest) returns (ExampleResponse);
  rpc ExampleUnaryStream(ExampleRequest) returns (stream ExampleResponse);
  rpc ExampleStreamUnary(stream ExampleRequest) returns (ExampleResponse);
  rpc ExampleStreamStream(stream ExampleRequest) returns (stream ExampleResponse);
}

message ExampleRequest {
  string example_string = 1;
  int64  example_integer = 2;
}

message ExampleResponse {
  string example_string = 1;
  int64  example_integer = 2;
}

Using with protoc

Prerequisites

Make sure you have:

• protoc (Protocol Buffer compiler) installed
• The protoc-slimrpc-plugin binary in your PATH or specify its full path

Generate Python Files

# Generate both the protobuf Python files and SLIMRPC files
protoc \
  --python_out=. \
  --pyi_out=. \
  --plugin=protoc-gen-slimrpc=${HOME}/.cargo/bin/protoc-slimrpc-plugin \
  --slimrpc_out=. \
  example.proto

This will generate:

• example_pb2.py - Standard protobuf Python bindings
• example_pb2_slimrpc.py - SLIMRPC client stubs and server servicers

With Custom Types Import

You can specify a custom import for the types module. This allows to import the types from an external package.

For instance, if you don't want to generate the types and you want to import them from a2a.grpc.a2a_pb2`, you can do:

protoc \
  --plugin=protoc-gen-slimrpc=${HOME}/.cargo/bin/protoc-slimrpc-plugin \
  --slimrpc_out=types_import="from a2a.grpc import a2a_pb2 as a2a__pb2":. \
  example.proto

Using with buf

Prerequisites

• buf CLI installed
• protoc-slimrpc-plugin binary in your PATH, or specify the full path in the buf.gen.yaml file

Create buf.gen.yaml

Create a buf.gen.yaml file in your project root:

version: v2
managed:
  enabled: true
inputs:
  - proto_file: example.proto
plugins:
  - local: /path/to/protoc-slimrpc-plugin
    out: .
  - remote: buf.build/protocolbuffers/python
    out: .

Generate Code

buf generate

Advanced buf Configuration

As before, you can customize the types import. For example, to use existing types from a2a.grpc.a2a_pb2, you can modify the buf.gen.yaml as follows:

version: v2
managed:
  enabled: true
plugins:
  - local: protoc-slimrpc-plugin
    out: generated
    opt:
      - types_import=from a2a.grpc import a2a_pb2 as a2a__pb2
  - remote: buf.build/protocolbuffers/python
    out: generated

Generated Code Structure

For the example above, the generated example_pb2_slimrpc.py will contain:

Client Stub

class TestStub:
    """Client stub for Test."""
    def __init__(self, channel):
        """Constructor.

        Args:
            channel: A slimrpc.Channel.
        """
        self.ExampleUnaryUnary = channel.unary_unary(...)
        self.ExampleUnaryStream = channel.unary_stream(...)
        # ... other methods

Server Servicer

class TestServicer():
    """Server servicer for Test. Implement this class to provide your service logic."""

    def ExampleUnaryUnary(self, request, context):
        """Method for ExampleUnaryUnary. Implement your service logic here."""
        raise slimrpc_rpc.SLIMRPCResponseError(
            code=code__pb2.UNIMPLEMENTED, message="Method not implemented!"
        )
    # ... other methods

Registration Function

def add_TestServicer_to_server(servicer, server: slimrpc.Server):
    # Registers the servicer with the SLIMRPC server
    pass

Plugin Parameters

The plugin supports the following parameters:

• types_import: Customize how protobuf types are imported
• Example: types_import="from my_package import types_pb2 as pb2"
• Default: Uses local import based on the proto file name

Example Usage in Python

Client Usage

import asyncio
import logging
from collections.abc import AsyncGenerator

import slimrpc
from slimrpc.examples.simple.types.example_pb2 import ExampleRequest
from slimrpc.examples.simple.types.example_pb2_slimrpc import TestStub

logger = logging.getLogger(__name__)


async def amain() -> None:
    channel = slimrpc.Channel(
        local="agntcy/grpc/client",
        slim={
            "endpoint": "http://localhost:46357",
            "tls": {
                "insecure": True,
            },
        },
        enable_opentelemetry=False,
        shared_secret="my_shared_secret",
        remote="agntcy/grpc/server",
    )

    # Stubs
    stubs = TestStub(channel)

    # Call method
    try:
        request = ExampleRequest(example_integer=1, example_string="hello")
        response = await stubs.ExampleUnaryUnary(request, timeout=2)

        logger.info(f"Response: {response}")

        responses = stubs.ExampleUnaryStream(request, timeout=2)
        async for resp in responses:
            logger.info(f"Stream Response: {resp}")

        async def stream_requests() -> AsyncGenerator[ExampleRequest]:
            for i in range(10):
                yield ExampleRequest(example_integer=i, example_string=f"Request {i}")

        response = await stubs.ExampleStreamUnary(stream_requests(), timeout=2)
        logger.info(f"Stream Unary Response: {response}")
    except asyncio.TimeoutError:
        logger.error("timeout while waiting for response")

Server Usage

import asyncio
import logging
from collections.abc import AsyncIterable

from slimrpc.context import Context
from slimrpc.examples.simple.types.example_pb2 import ExampleRequest, ExampleResponse
from slimrpc.examples.simple.types.example_pb2_slimrpc import (
    TestServicer,
    add_TestServicer_to_server,
)
from slimrpc.server import Server

logger = logging.getLogger(__name__)


class TestService(TestServicer):
    async def ExampleUnaryUnary(
        self, request: ExampleRequest, context: Context
    ) -> ExampleResponse:
        logger.info(f"Received unary-unary request: {request}")

        return ExampleResponse(example_integer=1, example_string="Hello, World!")

    async def ExampleUnaryStream(
        self, request: ExampleRequest, context: Context
    ) -> AsyncIterable[ExampleResponse]:
        logger.info(f"Received unary-stream request: {request}")

        # generate async responses stream
        for i in range(5):
            logger.info(f"Sending response {i}")
            yield ExampleResponse(example_integer=i, example_string=f"Response {i}")

    async def ExampleStreamUnary(
        self, request_iterator: AsyncIterable[ExampleRequest], context: Context
    ) -> ExampleResponse:
        logger.info(f"Received stream-unary request: {request_iterator}")

        async for request in request_iterator:
            logger.info(f"Received stream-unary request: {request}")
        response = ExampleResponse(
            example_integer=1, example_string="Stream Unary Response"
        )
        return response

    async def ExampleStreamStream(
        self, request_iterator: AsyncIterable[ExampleRequest], context: Context
    ) -> AsyncIterable[ExampleResponse]:
        """Missing associated documentation comment in .proto file."""
        raise NotImplementedError("Method not implemented!")


def create_server(
    local: str,
    slim: dict,
    enable_opentelemetry: bool = False,
    shared_secret: str = "",
) -> Server:
    """
    Create a new SLIMRPC server instance.
    """
    server = Server(
        local=local,
        slim=slim,
        enable_opentelemetry=enable_opentelemetry,
        shared_secret=shared_secret,
    )

    return server


async def amain() -> None:
    server = create_server(
        local="agntcy/grpc/server",
        slim={
            "endpoint": "http://localhost:46357",
            "tls": {
                "insecure": True,
            },
        },
        enable_opentelemetry=False,
        shared_secret="my_shared_secret",
    )

    # Create RPCs
    add_TestServicer_to_server(
        TestService(),
        server,
    )

    await server.run()

Troubleshooting

Plugin Not Found

If you get an error that the plugin is not found:

• Ensure protoc-slimrpc-plugin is in your PATH
• Or specify the full path: --plugin=protoc-gen-slimrpc=/full/path/to/protoc-slimrpc-plugin

Import Errors

If you encounter Python import errors:

• Make sure the generated *_pb2.py files are in your Python path.
• Use the types_import parameter to customize import paths.
• Ensure all Protocol Buffer dependencies are generated.

Build Errors

If the plugin fails to build:

• Ensure you have Rust and Cargo installed.
• Check that all dependencies are available.
• Try cleaning and rebuilding: cargo clean && cargo build --release.