Auth & Context

Authentication and request-scoped context for RPC methods.

Quick Overview

Server methods can accept an optional ctx: CallContext parameter. The framework injects it automatically — it does not appear in the Protocol definition:

from vgi_rpc import CallContext


class MyServiceImpl:
    def public_method(self) -> str:
        """No ctx — accessible to all callers."""
        return "ok"

    def protected_method(self, ctx: CallContext) -> str:
        """Require authentication, then return caller identity."""
        ctx.auth.require_authenticated()
        return f"Hello, {ctx.auth.principal}"

HTTP authentication

Pass an authenticate callback to make_wsgi_app:

import falcon

from vgi_rpc import AuthContext, RpcServer, make_wsgi_app


def authenticate(req: falcon.Request) -> AuthContext:
    token = req.get_header("Authorization") or ""
    if not token.startswith("Bearer "):
        raise ValueError("Missing Bearer token")
    # ... validate token ...
    return AuthContext(domain="jwt", authenticated=True, principal="alice")


server = RpcServer(MyService, MyServiceImpl())
app = make_wsgi_app(server, authenticate=authenticate)

Over pipe/subprocess transport, ctx.auth is always AuthContext.anonymous().

Transport metadata

ctx.transport_metadata provides transport-level information like remote_addr and user_agent (HTTP only). It's a read-only mapping populated by the transport layer.

API Reference

AuthContext

AuthContext dataclass

AuthContext(
    domain: str | None,
    authenticated: bool,
    principal: str | None = None,
    claims: Mapping[str, Any] = dict(),
)

Authentication context for the current request.

Populated by transport-specific authenticate callbacks (e.g. JWT validation on HTTP) and injected into method implementations via :class:CallContext.

ATTRIBUTE	DESCRIPTION
domain	Authentication scheme that produced this context, or None for unauthenticated requests. TYPE: str | None
authenticated	Whether the caller was successfully authenticated. TYPE: bool
principal	Identity of the caller (e.g. username, service account). TYPE: str | None
claims	Arbitrary claims from the authentication token. TYPE: Mapping[str, Any]

anonymous classmethod

anonymous() -> AuthContext

Unauthenticated context (default for pipe transport).

Source code in vgi_rpc/rpc/_common.py

@classmethod
def anonymous(cls) -> AuthContext:
    """Unauthenticated context (default for pipe transport)."""
    return _ANONYMOUS

require_authenticated

require_authenticated() -> None

Raise if not authenticated.

RAISES	DESCRIPTION
PermissionError	If authenticated is False.

Source code in vgi_rpc/rpc/_common.py

def require_authenticated(self) -> None:
    """Raise if not authenticated.

    Raises:
        PermissionError: If ``authenticated`` is ``False``.

    """
    if not self.authenticated:
        raise PermissionError("Authentication required")

CallContext

CallContext

CallContext(
    auth: AuthContext,
    emit_client_log: ClientLog,
    transport_metadata: Mapping[str, Any] | None = None,
    *,
    server_id: str = "",
    method_name: str = "",
    protocol_name: str = ""
)

Request-scoped context injected into methods that declare a ctx parameter.

Provides authentication, logging, and transport metadata in a single injection point.

Initialize with auth context, client-log callback, and optional server context fields.

Source code in vgi_rpc/rpc/_common.py

def __init__(
    self,
    auth: AuthContext,
    emit_client_log: ClientLog,
    transport_metadata: Mapping[str, Any] | None = None,
    *,
    server_id: str = "",
    method_name: str = "",
    protocol_name: str = "",
) -> None:
    """Initialize with auth context, client-log callback, and optional server context fields."""
    self.auth = auth
    self.emit_client_log = emit_client_log
    self.transport_metadata: Mapping[str, Any] = transport_metadata or {}
    self._server_id = server_id
    self._method_name = method_name
    self._protocol_name = protocol_name
    self._request_id = _current_request_id.get()
    self._logger: _ContextLoggerAdapter | None = None

request_id property

request_id: str

Per-request correlation ID (empty string if not set).

logger property

logger: LoggerAdapter[Logger]

Server-side logger with request context pre-bound.

RETURNS	DESCRIPTION
LoggerAdapter[Logger]	A LoggerAdapter with logger name
LoggerAdapter[Logger]	vgi_rpc.service.<ProtocolName>. Always includes
LoggerAdapter[Logger]	server_id and method; conditionally includes
LoggerAdapter[Logger]	request_id, principal, auth_domain, and
LoggerAdapter[Logger]	remote_addr when available.

client_log

client_log(
    level: Level, message: str, **extra: str
) -> None

Emit a client-directed log message (convenience wrapper).

Source code in vgi_rpc/rpc/_common.py

def client_log(self, level: Level, message: str, **extra: str) -> None:
    """Emit a client-directed log message (convenience wrapper)."""
    self.emit_client_log(Message(level, message, **extra))

CallStatistics

CallStatistics dataclass

CallStatistics(
    input_batches: int = 0,
    output_batches: int = 0,
    input_rows: int = 0,
    output_rows: int = 0,
    input_bytes: int = 0,
    output_bytes: int = 0,
)

Mutable accumulator of per-call I/O counters for usage accounting.

Created at dispatch start and populated as batches flow through the server. Surfaced through the access log and OTel dispatch hook.

Byte measurement: uses pa.RecordBatch.get_total_buffer_size() which reports logical Arrow buffer sizes (O(columns), negligible cost). This is an approximation — it does not include IPC framing overhead (padding, schema messages, EOS markers).

ATTRIBUTE	DESCRIPTION
input_batches	Number of input batches read by the server. TYPE: int
output_batches	Number of output batches written by the server. TYPE: int
input_rows	Total rows across all input batches. TYPE: int
output_rows	Total rows across all output batches. TYPE: int
input_bytes	Approximate logical bytes across all input batches. TYPE: int
output_bytes	Approximate logical bytes across all output batches. TYPE: int

record_input

record_input(batch: RecordBatch) -> None

Record an input batch's row count and buffer size.

Source code in vgi_rpc/rpc/_common.py

def record_input(self, batch: pa.RecordBatch) -> None:
    """Record an input batch's row count and buffer size."""
    self.input_batches += 1
    self.input_rows += batch.num_rows
    self.input_bytes += batch.get_total_buffer_size()

record_output

record_output(batch: RecordBatch) -> None

Record an output batch's row count and buffer size.

Source code in vgi_rpc/rpc/_common.py

def record_output(self, batch: pa.RecordBatch) -> None:
    """Record an output batch's row count and buffer size."""
    self.output_batches += 1
    self.output_rows += batch.num_rows
    self.output_bytes += batch.get_total_buffer_size()

ClientLog

ClientLog module-attribute

ClientLog = Callable[[Message], None]

Callback type for emitting client-directed log messages from RPC method implementations.