blazerpc 1.1.0

A lightweight, framework-agnostic gRPC library for machine learning inference

BlazeRPC

A lightweight, framework-agnostic gRPC library for serving machine learning models in Python. BlazeRPC gives you a FastAPI-like developer experience -- decorate a function, start the server, and you have a production-ready gRPC inference endpoint.

Why BlazeRPC?

Serving ML models over gRPC typically involves writing .proto files by hand, compiling them into Python stubs, and wiring up boilerplate servicers. BlazeRPC removes all of that. You write a plain Python function, add a decorator, and the library generates the protobuf schema, servicer, and server for you.

Key features:

• Decorator-based API -- Register models with @app.model("name"), just like route handlers in a web framework.
• Automatic proto generation -- BlazeRPC inspects your function's type annotations and produces a valid .proto file. No hand-written schemas.
• Adaptive batching -- Individual requests are automatically grouped into batches for GPU-efficient inference. Configurable batch size and timeout.
• Server-side streaming -- Return tokens one at a time with streaming=True, ideal for LLM inference and real-time pipelines.
• Health checks and reflection -- Built-in gRPC health checking protocol and server reflection, compatible with grpcurl, grpcui, and Kubernetes probes.
• Framework integrations -- Optional helpers for PyTorch, TensorFlow, and ONNX Runtime that handle tensor conversion automatically.
• Prometheus metrics -- Request counts, latencies, and batch sizes are exported out of the box.

Installation

pip install blazerpc

With framework-specific extras:

pip install blazerpc[pytorch]      # PyTorch tensor conversion helpers
pip install blazerpc[tensorflow]   # TensorFlow tensor conversion helpers
pip install blazerpc[onnx]         # ONNX Runtime model wrapper
pip install blazerpc[all]          # All optional integrations

Quick start

1. Define your models

Create a file called app.py:

from blazerpc import BlazeApp

app = BlazeApp()

@app.model("sentiment")
def predict_sentiment(text: list[str]) -> list[float]:
    # Replace with your real model inference
    return [0.95] * len(text)

BlazeRPC reads the type annotations on your function to generate the gRPC request and response messages. Supported types include str, int, float, bool, list[float], list[str], and tensor types via TensorInput / TensorOutput.

2. Start the server

blaze serve app:app

⚡ BlazeRPC server starting...
  ✓ Loaded model: sentiment v1
  ✓ Server listening on 0.0.0.0:50051

The server registers three services automatically:

Service	Purpose
blazerpc.InferenceService	Your model RPCs
grpc.health.v1.Health	Standard health checks
grpc.reflection.v1alpha.ServerReflection	Service discovery

3. Export the .proto file

blaze proto app:app --output-dir ./proto_out

This writes a blaze_service.proto file that you can compile with protoc or share with clients in any language. The generated proto looks like this:

syntax = "proto3";
package blazerpc;

message TensorProto {
  repeated int64 shape = 1;
  string dtype = 2;
  bytes data = 3;
}

message SentimentRequest {
  repeated string text = 1;
}

message SentimentResponse {
  repeated float result = 1;
}

service InferenceService {
  rpc PredictSentiment(SentimentRequest) returns (SentimentResponse);
}

Streaming

To build a server-streaming endpoint (for example, returning tokens from an LLM), set streaming=True:

@app.model("generate", streaming=True)
async def generate_tokens(prompt: str) -> str:
    tokens = run_my_llm(prompt)
    for token in tokens:
        yield token

Each yield sends a message to the client over the open gRPC stream. The client receives tokens as they are produced, without waiting for the full response.

Adaptive batching

When enable_batching=True (the default), BlazeRPC collects individual requests and groups them into batches before calling your model function. This is essential for GPU workloads where batch inference is significantly faster than processing requests one at a time.

app = BlazeApp(
    enable_batching=True,
    max_batch_size=32,       # Maximum requests per batch
    batch_timeout_ms=10.0,   # Maximum wait time before dispatching a partial batch
)

The batching layer handles:

• Collecting requests from concurrent clients into a single batch.
• Dispatching partial batches when the timeout expires, ensuring low latency even under light load.
• Partial failure isolation -- if one item in a batch fails, only that client receives an error. Other clients in the batch still get their results.

Tensor types

For models that operate on NumPy arrays, use TensorInput and TensorOutput to declare the expected shape and dtype:

import numpy as np
from blazerpc import BlazeApp, TensorInput, TensorOutput

app = BlazeApp()

@app.model("classify")
def classify(
    image: TensorInput[np.float32, "batch", 224, 224, 3],
) -> TensorOutput[np.float32, "batch", 1000]:
    # image is serialized as a TensorProto on the wire
    return model.predict(image)

The generated proto uses a TensorProto message with shape, dtype, and raw bytes fields for zero-copy serialization.

Framework integrations

PyTorch

from blazerpc.contrib.pytorch import torch_model

@app.model("classifier")
@torch_model(device="cuda")
def classify(image):
    # `image` is automatically converted from np.ndarray to a torch.Tensor
    # on the specified device. The return value is converted back to np.ndarray.
    return model(image)

TensorFlow

from blazerpc.contrib.tensorflow import tf_model

@app.model("classifier")
@tf_model
def classify(image):
    return model(image)

ONNX Runtime

from blazerpc.contrib.onnx import ONNXModel

onnx_model = ONNXModel("model.onnx", providers=["CUDAExecutionProvider"])

@app.model("classifier")
def classify(image: np.ndarray) -> np.ndarray:
    return onnx_model.predict(image)[0]

Middleware

BlazeRPC provides a middleware system built on grpclib's event hooks. Attach middleware to the underlying server to add logging, metrics, or custom request processing.

from blazerpc.server.middleware import LoggingMiddleware, MetricsMiddleware

# These are attached inside app.serve() or manually on the grpclib Server:
# LoggingMiddleware().attach(grpclib_server)
# MetricsMiddleware().attach(grpclib_server)

Built-in middleware:

Middleware	Description
LoggingMiddleware	Logs every RPC call with method name, peer address, and response status.
MetricsMiddleware	Exports Prometheus metrics: blazerpc_requests_total and blazerpc_request_duration_seconds.
ExceptionMiddleware	Base class for custom exception-to-gRPC-status mapping.

To build your own middleware, subclass Middleware and implement on_request and on_response:

from blazerpc.server.middleware import Middleware

class AuthMiddleware(Middleware):
    async def on_request(self, event):
        token = dict(event.metadata).get("authorization")
        if not token:
            raise GRPCError(Status.UNAUTHENTICATED, "Missing token")

    async def on_response(self, event):
        pass

CLI reference

blaze serve <app_path> [OPTIONS]

  Start the BlazeRPC gRPC server.

  Arguments:
    app_path    App import path in module:attribute format (e.g. app:app)

  Options:
    --host TEXT       Host to bind to              [default: 0.0.0.0]
    --port INTEGER    Port to listen on            [default: 50051]
    --workers INTEGER Number of worker processes   [default: 1]
    --reload          Enable auto-reload           [default: False]

blaze proto <app_path> [OPTIONS]

  Export the generated .proto file.

  Arguments:
    app_path    App import path in module:attribute format (e.g. app:app)

  Options:
    --output-dir TEXT  Output directory for .proto files  [default: .]

Project structure

src/blazerpc/
  __init__.py          # Public API: BlazeApp, TensorInput, TensorOutput, exceptions
  app.py               # BlazeApp class -- model registration and server lifecycle
  types.py             # TensorInput, TensorOutput, type introspection
  exceptions.py        # Exception hierarchy (BlazeRPCError and subclasses)
  decorators.py        # Reserved for future decorator extensions
  cli/
    main.py            # Typer CLI (blaze serve, blaze proto)
    serve.py           # App loading from import strings
    proto.py           # Proto file export
  codegen/
    proto.py           # .proto file generation from type annotations
    servicer.py        # Dynamic grpclib servicer generation
  runtime/
    registry.py        # Model registry (stores registered models and metadata)
    executor.py        # Model execution with sync/async bridging
    batcher.py         # Adaptive request batching
    serialization.py   # Tensor and scalar serialization
  server/
    grpc.py            # GRPCServer wrapper with signal handling and graceful shutdown
    health.py          # gRPC health checking protocol
    reflection.py      # gRPC server reflection
    middleware.py       # Logging, metrics, and extensible middleware base
  contrib/
    pytorch.py         # PyTorch <-> NumPy conversion and @torch_model decorator
    tensorflow.py      # TensorFlow <-> NumPy conversion and @tf_model decorator
    onnx.py            # ONNX Runtime session wrapper

Development

# Clone the repository
git clone https://github.com/Ifihan/blazerpc.git
cd blazerpc

# Install dependencies (requires uv)
uv sync --extra dev

# Run tests
uv run pytest tests/ -v

# Lint
uv run ruff check src/

# Type check
uv run mypy src/blazerpc/

Contributing

We welcome contributions of all kinds -- bug fixes, new features, documentation improvements, and example applications. See the Contributing Guide for instructions on setting up a development environment, running tests, and submitting a pull request.

License

MIT -- see LICENSE for details.