gRPC Client Code Generator and Tester

gRPC Client Code Generator and Tester

Problem Statement

Working with gRPC services during development and debugging is painful. Teams typically end up writing throwaway Python clients for every service, recompiling .proto files by hand, and juggling generated stubs across virtual environments. When a service changes, all those ad-hoc scripts break, and there's no easy way to sanity-check a deployed gRPC endpoint without spinning up a full client application.

This script solves that by being a single-file, self-contained gRPC introspection and testing tool. Given a .proto file and a server address, it dynamically compiles the proto, imports the generated stubs, resolves the requested RPC method, and invokes it with JSON payload data — all in one command. It's the gRPC equivalent of curl for REST APIs, and it's invaluable for CI smoke tests, on-call debugging, and API exploration.

How It Works

The script follows a four-stage pipeline:

1. Compile — It invokes grpc_tools.protoc programmatically against the provided .proto file, emitting *_pb2.py and *_pb2_grpc.py stubs into a temporary directory. This avoids polluting the working tree and makes the tool idempotent.

2. Load — It inserts the temp directory onto sys.path, then uses importlib to dynamically import the generated _pb2 (message definitions) and _pb2_grpc (service stubs) modules. Because imports are dynamic, the script doesn't need to know service names ahead of time.

3. Resolve — Using the reflection-like DESCRIPTOR objects on the generated modules, it walks the service and method descriptors to find the requested RPC. This gives us the input message type, so we can build a proper protobuf message from the JSON payload.

4. Invoke — It constructs a channel (with optional TLS), instantiates the stub, converts the JSON payload into the protobuf message via json_format.ParseDict, and calls the method. The response is serialized back to JSON via json_format.MessageToDict for human-readable output.

Design decisions worth noting: the script uses TemporaryDirectory for cleanup guarantees, argparse for a familiar CLI surface, and structured exit codes (0 success, 1 usage error, 2 RPC error) so it composes well in shell pipelines.

Full Script

#!/usr/bin/env python3
"""
gRPC Client Code Generator and Tester.

Compiles a .proto file on the fly, loads the generated stubs, and invokes
a named RPC method against a target server using a JSON payload.

Example:
    python3 grpc_client.py --proto service.proto \\
        --server localhost:50051 --method GetUser --data '{"id":1}'
"""

from __future__ import annotations

import argparse
import importlib
import json
import os
import sys
import tempfile
import traceback
from pathlib import Path
from typing import Any

import grpc
from grpc_tools import protoc
from google.protobuf import json_format


# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_TIMEOUT = float(os.environ.get("GRPC_TIMEOUT", "10.0"))
DEFAULT_TLS = os.environ.get("GRPC_TLS", "false").lower() in {"1", "true", "yes"}
DEFAULT_METADATA = os.environ.get("GRPC_METADATA", "")  # "k1=v1,k2=v2"
DEFAULT_VERBOSE = os.environ.get("GRPC_VERBOSE", "false").lower() in {"1", "true", "yes"}


# ---------------------------------------------------------------------------
# Proto compilation
# ---------------------------------------------------------------------------
def compile_proto(proto_path: Path, out_dir: Path) -> None:
    """Compile a .proto file into Python stubs inside out_dir."""
    proto_dir = proto_path.parent.resolve()
    include_dir = Path(protoc.__file__).parent / "_proto"

    args = [
        "protoc",
        f"--proto_path={proto_dir}",
        f"--proto_path={include_dir}",
        f"--python_out={out_dir}",
        f"--grpc_python_out={out_dir}",
        str(proto_path.resolve()),
    ]

    rc = protoc.main(args)
    if rc != 0:
        raise RuntimeError(f"protoc failed with exit code {rc} for {proto_path}")


# ---------------------------------------------------------------------------
# Dynamic module loading
# ---------------------------------------------------------------------------
def load_stub_modules(proto_path: Path, out_dir: Path):
    """Import the generated *_pb2 and *_pb2_grpc modules."""
    sys.path.insert(0, str(out_dir))
    base = proto_path.stem  # e.g. "service" from "service.proto"

    try:
        pb2 = importlib.import_module(f"{base}_pb2")
        pb2_grpc = importlib.import_module(f"{base}_pb2_grpc")
    except ImportError as exc:
        raise RuntimeError(
            f"Could not import generated modules for {base!r}: {exc}"
        ) from exc

    return pb2, pb2_grpc


# ---------------------------------------------------------------------------
# Service / method introspection
# ---------------------------------------------------------------------------
def find_service_and_method(pb2_grpc, pb2, method_name: str):
    """Locate a stub class and method matching method_name across services."""
    for attr_name in dir(pb2_grpc):
        attr = getattr(pb2_grpc, attr_name)
        if not isinstance(attr, type) or not hasattr(attr, "__init__"):
            continue
        # Stub classes end with "Stub"
        if not attr_name.endswith("Stub"):
            continue
        if hasattr(attr, method_name):
            # Find the matching service descriptor in pb2 for input type
            service_name = attr_name[: -len("Stub")]
            service_desc = pb2.DESCRIPTOR.services_by_name.get(service_name)
            if service_desc is None:
                continue
            method_desc = service_desc.methods_by_name.get(method_name)
            if method_desc is not None:
                return attr, method_desc
    raise LookupError(f"No service exposes RPC method {method_name!r}")


# ---------------------------------------------------------------------------
# Channel construction
# ---------------------------------------------------------------------------
def build_channel(server: str, use_tls: bool) -> grpc.Channel:
    if use_tls:
        credentials = grpc.ssl_channel_credentials()
        return grpc.secure_channel(server, credentials)
    return grpc.insecure_channel(server)


def parse_metadata(raw: str) -> list[tuple[str, str]]:
    if not raw:
        return []
    pairs: list[tuple[str, str]] = []
    for item in raw.split(","):
        if "=" not in item:
            raise ValueError(f"Malformed metadata entry: {item!r} (expected k=v)")
        k, v = item.split("=", 1)
        pairs.append((k.strip(), v.strip()))
    return pairs


# ---------------------------------------------------------------------------
# Invocation
# ---------------------------------------------------------------------------
def invoke(
    server: str,
    proto_path: Path,
    method_name: str,
    payload: dict[str, Any],
    timeout: float,
    use_tls: bool,
    metadata: list[tuple[str, str]],
    verbose: bool,
) -> dict[str, Any]:
    with tempfile.TemporaryDirectory(prefix="grpc_stubs_") as tmp:
        out_dir = Path(tmp)
        compile_proto(proto_path, out_dir)
        pb2, pb2_grpc = load_stub_modules(proto_path, out_dir)

        stub_cls, method_desc = find_service_and_method(pb2_grpc, pb2, method_name)
        if verbose:
            print(
                f"[verbose] Resolved {stub_cls.__name__}.{method_name} "
                f"(input={method_desc.input_type.full_name})",
                file=sys.stderr,
            )

        request = json_format.ParseDict(payload, method_desc.input_type._concrete_class)

        with build_channel(server, use_tls) as channel:
            stub = stub_cls(channel)
            rpc = getattr(stub, method_name)
            response = rpc(request, timeout=timeout, metadata=metadata or None)

        return json_format.MessageToDict(
            response, preserving_proto_field_name=True, including_default_value_fields=False
        )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Compile a .proto and invoke a gRPC method with JSON data.",
    )
    p.add_argument("--proto", required=True, help="Path to .proto file")
    p.add_argument("--server", required=True, help="host:port of gRPC server")
    p.add_argument("--method", required=True, help="RPC method name (e.g. GetUser)")
    p.add_argument("--data", default="{}", help="JSON payload for the request")
    p.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"RPC timeout in seconds (default {DEFAULT_TIMEOUT})",
    )
    p.add_argument(
        "--tls",
        action="store_true",
        default=DEFAULT_TLS,
        help="Use TLS for the channel",
    )
    p.add_argument(
        "--metadata",
        default=DEFAULT_METADATA,
        help="Comma-separated k=v metadata headers",
    )
    p.add_argument("-v", "--verbose", action="store_true", default=DEFAULT_VERBOSE)
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    proto_path = Path(args.proto)
    if not proto_path.is_file():
        print(f"error: proto file not found: {proto_path}", file=sys.stderr)
        return 1

    try:
        payload = json.loads(args.data)
        if not isinstance(payload, dict):
            raise ValueError("--data must decode to a JSON object")
    except (json.JSONDecodeError, ValueError) as exc:
        print(f"error: invalid --data JSON: {exc}", file=sys.stderr)
        return 1

    try:
        metadata = parse_metadata(args.metadata)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    try:
        result = invoke(
            server=args.server,
            proto_path=proto_path,
            method_name=args.method,
            payload=payload,
            timeout=args.timeout,
            use_tls=args.tls,
            metadata=metadata,
            verbose=args.verbose,
        )
    except LookupError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    except grpc.RpcError as exc:
        code = exc.code() if hasattr(exc, "code") else "UNKNOWN"
        details = exc.details() if hasattr(exc, "details") else str(exc)
        print(f"rpc error [{code}]: {details}", file=sys.stderr)
        if args.verbose:
            traceback.print_exc()
        return 2
    except (RuntimeError, ValueError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        if args.verbose:
            traceback.print_exc()
        return 1

    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

CLI flags mirror the env vars so the tool works both interactively and inside CI pipelines where env config is preferred.

Usage Examples

1. Plain unauthenticated call against a local server:

$ python3 grpc_client.py --proto service.proto \
    --server localhost:50051 --method GetUser --data '{"id": 1}'
{
  "email": "alice@example.com",
  "id": 1,
  "name": "Alice"
}

2. TLS + auth metadata against a staging server:

$ GRPC_TLS=true GRPC_METADATA="authorization=Bearer eyJhbGci..." \
  python3 grpc_client.py --proto service.proto \
    --server api.staging.example.com:443 \
    --method ListUsers --data '{"page_size": 10}'
{
  "users": [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
  ]
}

3. Verbose debugging with a bad method name:

$ python3 grpc_client.py --proto service.proto --server localhost:50051 \
    --method NoSuchMethod --data '{}' -v
[verbose] Resolved UserServiceStub.GetUser (input=example.UserRequest)
error: No service exposes RPC method 'NoSuchMethod'
$ echo $?
1

Error Handling

• Missing .proto file → checked with Path.is_file() before compilation; exits with code 1.
• protoc compilation failure → compile_proto checks the return code and raises RuntimeError; caught in main and reported.
• Malformed JSON payload → json.loads errors plus a type check that ensures the payload is an object; exits 1.
• Unknown RPC method → find_service_and_method raises LookupError with a clear message.
• Server-side gRPC errors (UNAVAILABLE, NOT_FOUND, PERMISSION_DENIED, etc.) → caught as grpc.RpcError, printing the status code and details; exits 2 so scripts can distinguish RPC failures from usage errors.
• Timeouts → handled by the --timeout deadline; surfaces as grpc.RpcError with code DEADLINE_EXCEEDED.
• Bad metadata format → parse_metadata raises ValueError before any network call.

Extending Ideas

• Server reflection — Add a mode that queries grpc.reflection.v1alpha.ServerReflection to fetch descriptors directly from the server, eliminating the need for a local .proto file entirely.
• Streaming support — Detect method_desc.client_streaming / server_streaming and iterate over request/response generators, reading NDJSON from stdin and writing NDJSON to stdout.
• Output formats — Add --output-format {json,yaml,table} so results can be piped into jq, yq, or rendered as a table for humans.
• Auth plugins — Abstract channel creation behind a plugin registry so users can plug in OAuth2, mTLS, or AWS SigV4 credential providers via a --auth flag.
• Pytest integration — Expose invoke() as a fixture-friendly function so teams can write assertion-based smoke tests against deployed gRPC services in their CI suite.