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)

vgi-rpc ships several built-in authenticate factories so you don't have to write the callback yourself:

Factory	Use case	Extra deps
`bearer_authenticate`	Opaque tokens / API keys with custom validation	None
`bearer_authenticate_static`	Fixed set of pre-shared tokens	None
`jwt_authenticate`	JWT validation against a JWKS endpoint	`vgi-rpc[oauth]`
`mtls_authenticate`	Client certificate with custom validation	`vgi-rpc[mtls]`
`mtls_authenticate_fingerprint`	Certificate fingerprint lookup	`vgi-rpc[mtls]`
`mtls_authenticate_subject`	Certificate Subject CN extraction	`vgi-rpc[mtls]`
`mtls_authenticate_xfcc`	Envoy XFCC header parsing	None
`chain_authenticate`	Compose multiple authenticators (e.g. JWT + mTLS + API key)	None

See OAuth Discovery for bearer/JWT details and Mutual TLS for mTLS details.

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.

cookies `property` ¶

cookies: Mapping[str, str]

Cookies from the incoming HTTP request.

Returns a read-only mapping of cookie name to value. Empty for non-HTTP transports (pipe, subprocess, shared memory) because those transports have no concept of HTTP cookies.

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))

set_cookie ¶

set_cookie(
    name: str,
    value: str,
    *,
    expires: datetime | None = None,
    max_age: int | None = None,
    domain: str | None = None,
    path: str | None = None,
    secure: bool | None = None,
    http_only: bool = True,
    same_site: str | None = None,
    partitioned: bool = False
) -> None

Queue a Set-Cookie header for the HTTP response.

Only valid inside a unary RPC method served over HTTP. Streaming responses are chunked with state carried in Arrow metadata and cannot cleanly attach a single Set-Cookie.

PARAMETER	DESCRIPTION
`name`	Cookie name. TYPE: `str`
`value`	Cookie value. TYPE: `str`
`expires`	Absolute expiry time. Mutually useful with `max_age`. TYPE: `datetime \| None` DEFAULT: `None`
`max_age`	Seconds until expiry. `0` deletes the cookie. TYPE: `int \| None` DEFAULT: `None`
`domain`	`Domain` attribute. TYPE: `str \| None` DEFAULT: `None`
`path`	`Path` attribute. TYPE: `str \| None` DEFAULT: `None`
`secure`	`Secure` attribute. `None` lets the transport decide. TYPE: `bool \| None` DEFAULT: `None`
`http_only`	`HttpOnly` attribute. Defaults to `True`. TYPE: `bool` DEFAULT: `True`
`same_site`	`SameSite` attribute (`"Strict"`, `"Lax"`, `"None"`). TYPE: `str \| None` DEFAULT: `None`
`partitioned`	`Partitioned` attribute (CHIPS). TYPE: `bool` DEFAULT: `False`

RAISES	DESCRIPTION
`RuntimeError`	If the call is not a unary HTTP request.

Source code in vgi_rpc/rpc/_common.py

def set_cookie(
    self,
    name: str,
    value: str,
    *,
    expires: datetime | None = None,
    max_age: int | None = None,
    domain: str | None = None,
    path: str | None = None,
    secure: bool | None = None,
    http_only: bool = True,
    same_site: str | None = None,
    partitioned: bool = False,
) -> None:
    """Queue a ``Set-Cookie`` header for the HTTP response.

    Only valid inside a unary RPC method served over HTTP.  Streaming
    responses are chunked with state carried in Arrow metadata and
    cannot cleanly attach a single ``Set-Cookie``.

    Args:
        name: Cookie name.
        value: Cookie value.
        expires: Absolute expiry time.  Mutually useful with ``max_age``.
        max_age: Seconds until expiry.  ``0`` deletes the cookie.
        domain: ``Domain`` attribute.
        path: ``Path`` attribute.
        secure: ``Secure`` attribute.  ``None`` lets the transport decide.
        http_only: ``HttpOnly`` attribute.  Defaults to ``True``.
        same_site: ``SameSite`` attribute (``"Strict"``, ``"Lax"``, ``"None"``).
        partitioned: ``Partitioned`` attribute (CHIPS).

    Raises:
        RuntimeError: If the call is not a unary HTTP request.

    """
    sink = _current_response_cookies.get()
    if sink is None:
        raise RuntimeError(
            "set_cookie() is only supported inside unary RPC methods served over HTTP",
        )
    sink.append(
        CookieSpec(
            name=name,
            value=value,
            expires=expires,
            max_age=max_age,
            domain=domain,
            path=path,
            secure=secure,
            http_only=http_only,
            same_site=same_site,
            partitioned=partitioned,
        )
    )

delete_cookie ¶

delete_cookie(
    name: str,
    *,
    path: str | None = None,
    domain: str | None = None
) -> None

Queue a cookie deletion on the HTTP response.

PARAMETER	DESCRIPTION
`name`	Cookie name to unset. TYPE: `str`
`path`	`Path` attribute of the cookie to match. TYPE: `str \| None` DEFAULT: `None`
`domain`	`Domain` attribute of the cookie to match. TYPE: `str \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`RuntimeError`	If the call is not a unary HTTP request.

Source code in vgi_rpc/rpc/_common.py

def delete_cookie(
    self,
    name: str,
    *,
    path: str | None = None,
    domain: str | None = None,
) -> None:
    """Queue a cookie deletion on the HTTP response.

    Args:
        name: Cookie name to unset.
        path: ``Path`` attribute of the cookie to match.
        domain: ``Domain`` attribute of the cookie to match.

    Raises:
        RuntimeError: If the call is not a unary HTTP request.

    """
    sink = _current_response_cookies.get()
    if sink is None:
        raise RuntimeError(
            "delete_cookie() is only supported inside unary RPC methods served over HTTP",
        )
    sink.append(CookieSpec(name=name, delete=True, path=path, domain=domain))

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.

Auth & Context¶

Quick Overview¶

HTTP authentication¶

Transport metadata¶

API Reference¶

AuthContext¶

AuthContext dataclass ¶

anonymous classmethod ¶

require_authenticated ¶

CallContext¶

CallContext ¶

request_id property ¶

logger property ¶

cookies property ¶

client_log ¶

set_cookie ¶

delete_cookie ¶

CallStatistics¶

CallStatistics dataclass ¶

record_input ¶

record_output ¶

ClientLog¶

ClientLog module-attribute ¶

AuthContext `dataclass` ¶

anonymous `classmethod` ¶

request_id `property` ¶

logger `property` ¶

cookies `property` ¶

CallStatistics `dataclass` ¶

ClientLog `module-attribute` ¶