Introspection

Discover an RPC service's methods, schemas, and parameter types at runtime or statically.

Usage

Static introspection (no server needed)

Use rpc_methods() to inspect a Protocol class, or describe_rpc() for a human-readable summary:

from vgi_rpc import describe_rpc, rpc_methods

methods = rpc_methods(Calculator)
for name, info in methods.items():
    print(f"{name}: {info.method_type.value}, params={info.params_schema}")

print(describe_rpc(Calculator))

Runtime introspection (over a connection you already hold)

An RpcServer built with enable_describe=True hosts the built-in vgi_rpc.Reflection.v1 protocol beside its application protocols. Ask it through the proxy you already have — on any transport — with list_protocols() and describe_protocol():

from vgi_rpc import connect, describe_protocol, list_protocols

with connect(Calculator, ["python", "worker.py"]) as proxy:
    for p in list_protocols(proxy):
        print(p.name, p.version, p.hash)
    desc = describe_protocol(proxy, "Calculator")
    for name, method in desc.methods.items():
        print(f"{name}: {method.method_type.value}")
    proxy.add(a=1.0, b=2.0)  # the connection is still yours

Both reuse the target's connection rather than opening one. A proxy from connect, serve_pipe, unix_connect, tcp_connect, iroh_connect, WorkerPool.connect or RpcConnection is rebound to reflection over the same byte stream — the server routes each request by its vgi_rpc.protocol key — so do not call them while a stream is open on that connection. A proxy from http_connect or httpi_connect shares its httpx2.Client, prefix, auth, retry and response-budget settings. The proxy may be bound to any protocol the server hosts; only its connection matters. A raw RpcTransport is accepted too.

list_protocols() returns one HostedProtocol per hosted protocol, in the server's order: application protocols in registration order (the primary first), then the framework's own (vgi_rpc.Reflection.v1, and vgi_rpc.Identity.v1 on an HTTP server that hosts it). hash is the protocol's canonical SHA-256, identical in every port for the same wire surface, so a client that cached a description for that hash can skip describe_protocol(). Use the listing to discover an optional protocol before calling it, rather than calling it and reading an error.

describe_protocol() makes two calls — list_protocols and then describe(name) — and returns a ServiceDescription. A name the server does not host is an RpcError with error_kind == "protocol_not_supported".

Servers without reflection. A server built without enable_describe=True (the default), or one older than reflection, answers "not hosted" (protocol_not_supported, an unknown method, UNIMPLEMENTED, or over HTTP a bare 404). Both functions raise ReflectionNotSupportedError for that answer — a subclass of RpcError carrying the server's error fields — and nothing else: no listing is inferred, because only the caller knows which protocol it expected the server to speak. The connection remains usable. Any other failure propagates as itself.

introspect(transport) and http_introspect(url) remain for callers that have only a raw transport or a URL and want the primary protocol described in one call:

from vgi_rpc import http_introspect

desc = http_introspect("http://localhost:8080")

The ServiceDescription carries language-neutral method metadata — name, method type, request/result/header Arrow schemas, and stream flags — everything a dynamic client needs to invoke methods without the Python Protocol class. It also includes protocol_hash (a SHA-256 digest identifying the contract) and protocol_version.

Discovery takes two round trips — list_protocols to learn what the server hosts, then describe for one of them. With a server free to host several protocols there is no longer a single "the" protocol to describe without first asking; both introspect() and http_introspect() accept an explicit protocol to skip the discovery hop.

The retired __describe__ method is gone: introspection is the ordinary vgi_rpc.Reflection.v1 protocol, dispatched through a normal binding, so it gets access logging, telemetry and the normal error envelope like any other call. A server that externalizes payloads externalizes reflection replies too, so an HTTP client needs its external_location configured to read them.

Python-flavoured fields (parameter type names, default values, docstrings) are not on the wire, having been dropped for cross-language neutrality. Consumers that need those human-readable details import the Protocol source class directly rather than reconstructing them from the wire. DESCRIBE_VERSION is reported as "5" and is vestigial — the protocol's major version is part of its own name now, so there is no separate format number to negotiate.

For stream methods that declare a header type (Stream[S, H]), MethodDescription includes has_header (bool) and header_schema (pa.Schema | None) fields describing the header's Arrow schema. Streams also carry is_exchange (bool | None) — True for exchange (bidi), False for producer, None for unary. Header fields are False / None for methods without headers.

API Reference

Functions

list_protocols

list_protocols(target: object) -> list[HostedProtocol]

List the protocols a server hosts, over a connection the caller holds.

One round trip -- vgi_rpc.Reflection.v1.list_protocols -- on target's own connection; nothing new is opened and nothing is closed. Over HTTP the call shares the proxy's httpx2.Client, prefix, auth, retry and response-budget settings; over every other transport it shares the proxy's byte stream, which the server demultiplexes by each request's protocol key. On a pipe-like transport, do not call it while a stream is open on the same connection: the two would interleave on one channel.

PARAMETER	DESCRIPTION
target	A proxy from any connect function -- connect, serve_pipe, unix_connect, tcp_connect, iroh_connect, http_connect, httpi_connect, WorkerPool.connect or RpcConnection -- bound to any protocol the server hosts, or a raw RpcTransport. TYPE: object

RETURNS	DESCRIPTION
One	class:HostedProtocol per hosted protocol, in the server's TYPE: list[HostedProtocol]
order	application protocols first, primary leading, then the TYPE: list[HostedProtocol]
list[HostedProtocol]	framework's own.

RAISES	DESCRIPTION
ReflectionNotSupportedError	The server does not host reflection (built without enable_describe=True, or older than it). The connection is still usable.
RpcError	The server answered with any other error, or the transport failed.
TypeError	target is neither a proxy nor an RpcTransport.

Source code in vgi_rpc/introspect.py

def list_protocols(target: object) -> list[HostedProtocol]:
    """List the protocols a server hosts, over a connection the caller holds.

    One round trip -- ``vgi_rpc.Reflection.v1.list_protocols`` -- on *target*'s
    own connection; nothing new is opened and nothing is closed.  Over HTTP
    the call shares the proxy's ``httpx2.Client``, prefix, auth, retry and
    response-budget settings; over every other transport it shares the
    proxy's byte stream, which the server demultiplexes by each request's
    protocol key.  On a pipe-like transport, do not call it while a stream is
    open on the same connection: the two would interleave on one channel.

    Args:
        target: A proxy from any connect function -- ``connect``,
            ``serve_pipe``, ``unix_connect``, ``tcp_connect``,
            ``iroh_connect``, ``http_connect``, ``httpi_connect``,
            ``WorkerPool.connect`` or ``RpcConnection`` -- bound to any
            protocol the server hosts, or a raw ``RpcTransport``.

    Returns:
        One :class:`HostedProtocol` per hosted protocol, in the server's
        order: application protocols first, primary leading, then the
        framework's own.

    Raises:
        ReflectionNotSupportedError: The server does not host reflection
            (built without ``enable_describe=True``, or older than it).
            The connection is still usable.
        RpcError: The server answered with any other error, or the
            transport failed.
        TypeError: *target* is neither a proxy nor an ``RpcTransport``.

    """
    listing = _list(_reflection_proxy(target))
    return [
        HostedProtocol(
            name=p.protocol,
            version=p.protocol_version,
            hash=p.protocol_hash,
            deprecated=p.deprecated,
            deprecation_message=p.deprecation_message,
            features=tuple(p.features),
        )
        for p in listing.protocols
    ]

describe_protocol

describe_protocol(
    target: object, name: str
) -> ServiceDescription

Describe one hosted protocol, over a connection the caller holds.

Two round trips on target's connection: list_protocols (for the server identity the description carries, and to tell "no reflection" apart from "no such protocol") then describe(name). The connection rules are those of :func:list_protocols.

PARAMETER	DESCRIPTION
target	A proxy from any connect function, or a raw RpcTransport; see :func:list_protocols. TYPE: object
name	The protocol's wire name, as :func:list_protocols reports it. TYPE: str

RETURNS	DESCRIPTION
A	class:ServiceDescription with each method's type and schemas. TYPE: ServiceDescription

RAISES	DESCRIPTION
ReflectionNotSupportedError	The server does not host reflection.
RpcError	The server does not host name (error_kind "protocol_not_supported"), answered with another error, or the transport failed.

Source code in vgi_rpc/introspect.py

def describe_protocol(target: object, name: str) -> ServiceDescription:
    """Describe one hosted protocol, over a connection the caller holds.

    Two round trips on *target*'s connection: ``list_protocols`` (for the
    server identity the description carries, and to tell "no reflection"
    apart from "no such protocol") then ``describe(name)``.  The connection
    rules are those of :func:`list_protocols`.

    Args:
        target: A proxy from any connect function, or a raw
            ``RpcTransport``; see :func:`list_protocols`.
        name: The protocol's wire name, as :func:`list_protocols` reports it.

    Returns:
        A :class:`ServiceDescription` with each method's type and schemas.

    Raises:
        ReflectionNotSupportedError: The server does not host reflection.
        RpcError: The server does not host *name* (``error_kind``
            ``"protocol_not_supported"``), answered with another error, or
            the transport failed.

    """
    proxy = _reflection_proxy(target)
    listing = _list(proxy)
    described: _WireDescription = proxy.describe(protocol=name)
    return _adapt_description(described, listing)

introspect

introspect(
    transport: RpcTransport,
    ipc_validation: IpcValidation = FULL,
    protocol: str | None = None,
) -> ServiceDescription

Describe a server's protocol over any RpcTransport.

Two round trips: list_protocols to learn what is hosted, then describe on one of them. The first is unavoidable now that a server may host several protocols -- there is no longer a single "the" protocol to ask about without asking.

PARAMETER	DESCRIPTION
transport	An open RpcTransport. TYPE: RpcTransport
ipc_validation	Validation level for incoming IPC batches. TYPE: IpcValidation DEFAULT: FULL
protocol	Which protocol to describe. Defaults to the first hosted one that is not reflection itself, which is the primary. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
ServiceDescription	A ServiceDescription with all method metadata.

RAISES	DESCRIPTION
RpcError	If the server does not support reflection or returns an error.
ValueError	If the server hosts no application protocol.

Source code in vgi_rpc/introspect.py

def introspect(
    transport: RpcTransport,
    ipc_validation: IpcValidation = IpcValidation.FULL,
    protocol: str | None = None,
) -> ServiceDescription:
    """Describe a server's protocol over any ``RpcTransport``.

    Two round trips: ``list_protocols`` to learn what is hosted, then
    ``describe`` on one of them.  The first is unavoidable now that a server
    may host several protocols -- there is no longer a single "the" protocol to
    ask about without asking.

    Args:
        transport: An open ``RpcTransport``.
        ipc_validation: Validation level for incoming IPC batches.
        protocol: Which protocol to describe.  Defaults to the first hosted
            one that is not reflection itself, which is the primary.

    Returns:
        A ``ServiceDescription`` with all method metadata.

    Raises:
        RpcError: If the server does not support reflection or returns an
            error.
        ValueError: If the server hosts no application protocol.

    """
    from vgi_rpc.rpc._reflection import ProtocolList
    from vgi_rpc.rpc._reflection import ServiceDescription as WireDescription

    listing = ProtocolList.deserialize_from_bytes(
        _reflection_call(transport, "list_protocols", _EMPTY_SCHEMA, {}, ipc_validation)
    )

    if protocol is None:
        application = [p for p in listing.protocols if not p.protocol.startswith("vgi_rpc.")]
        if not application:
            raise ValueError(f"Server {listing.server_id} hosts no application protocol.")
        protocol = application[0].protocol

    params_schema = pa.schema([pa.field("protocol", pa.utf8(), nullable=False)])
    described = WireDescription.deserialize_from_bytes(
        _reflection_call(transport, "describe", params_schema, {"protocol": protocol}, ipc_validation)
    )

    return _adapt_description(described, listing)

rpc_methods cached

rpc_methods(protocol: type) -> Mapping[str, RpcMethodInfo]

Introspect a Protocol class and return RpcMethodInfo for each method.

Skips underscore-prefixed names and non-callable attributes.

Source code in vgi_rpc/rpc/_types.py

@functools.lru_cache(maxsize=64)
def rpc_methods(protocol: type) -> Mapping[str, RpcMethodInfo]:
    """Introspect a Protocol class and return RpcMethodInfo for each method.

    Skips underscore-prefixed names and non-callable attributes.
    """
    result: dict[str, RpcMethodInfo] = {}

    # Get method names from Protocol — look at annotations and callables
    for name in dir(protocol):
        if name.startswith("_"):
            continue
        attr = getattr(protocol, name, None)
        if attr is None or not callable(attr):
            continue

        try:
            method_hints = get_type_hints(attr, include_extras=True)
        except (NameError, AttributeError) as exc:
            raise TypeError(f"Failed to resolve type hints for {protocol.__name__}.{name}(): {exc}") from exc

        _validate_protocol_params(protocol, name, inspect.signature(attr))

        return_hint = method_hints.get("return", type(None))
        method_type, result_type, has_return = _classify_return_type(return_hint)

        # For unary methods, build result schema from the result type
        result_schema = _build_result_schema(result_type) if method_type == MethodType.UNARY else _EMPTY_SCHEMA

        # Extract header type and stream sub-type from Stream[S, H] annotations
        header_type: type[ArrowSerializableDataclass] | None = None
        is_exchange: bool | None = None
        if method_type == MethodType.STREAM:
            stream_type, _, _ = _annotation_details(return_hint)
            stream_args = get_args(stream_type)
            if stream_args:
                s_arg = stream_args[0]
                if isinstance(s_arg, type):
                    if issubclass(s_arg, ExchangeState):
                        is_exchange = True
                    elif issubclass(s_arg, ProducerState):
                        is_exchange = False
            if len(stream_args) >= 2:
                h_arg = stream_args[1]
                if isinstance(h_arg, type) and issubclass(h_arg, ArrowSerializableDataclass):
                    header_type = h_arg

        param_defaults = _get_param_defaults(protocol, name)
        params_schema = _build_params_schema(method_hints)
        param_types = {k: v for k, v in method_hints.items() if k not in ("self", "return")}

        doc = getattr(attr, "__doc__", None)
        param_docs = _extract_param_docs(doc)

        result[name] = RpcMethodInfo(
            name=name,
            params_schema=params_schema,
            result_schema=result_schema,
            result_type=result_type,
            method_type=method_type,
            has_return=has_return,
            doc=doc,
            param_defaults=param_defaults,
            param_types=param_types,
            param_docs=param_docs,
            header_type=header_type,
            is_exchange=is_exchange,
            protocol_name=_protocol_wire_name(protocol),
        )

    return MappingProxyType(result)

describe_rpc

describe_rpc(
    protocol: type,
    *,
    methods: Mapping[str, RpcMethodInfo] | None = None
) -> str

Return a human-readable description of an RPC protocol's methods.

Source code in vgi_rpc/rpc/__init__.py

def describe_rpc(protocol: type, *, methods: Mapping[str, RpcMethodInfo] | None = None) -> str:
    """Return a human-readable description of an RPC protocol's methods."""
    if methods is None:
        methods = rpc_methods(protocol)
    lines: list[str] = [f"RPC Protocol: {protocol.__name__}", ""]

    for name, info in sorted(methods.items()):
        lines.append(f"  {name}({info.method_type.value})")
        lines.append(f"    params: {info.params_schema}")
        if info.method_type == MethodType.UNARY:
            lines.append(f"    result: {info.result_schema}")
        if info.doc:
            lines.append(f"    doc: {info.doc.strip()}")
        lines.append("")

    return "\n".join(lines)

Data Classes

HostedProtocol dataclass

HostedProtocol(
    name: str,
    version: str,
    hash: str,
    deprecated: bool = False,
    deprecation_message: str = "",
    features: tuple[str, ...] = (),
)

One protocol a server hosts, as vgi_rpc.Reflection.v1 lists it.

A client-side view of the wire ProtocolSummary, returned by :func:list_protocols in the server's order: application protocols in registration order (the primary first), then the framework's own (vgi_rpc.Reflection.v1, and vgi_rpc.Identity.v1 on an HTTP server that hosts it).

ATTRIBUTE	DESCRIPTION
name	The protocol's wire name -- its routing key, carrying its major version, e.g. "vgi_rpc.Reflection.v1". TYPE: str
version	Its declared semver, or "" when it declares none. TYPE: str
hash	SHA-256 of its canonical description, as 64 lowercase hex characters. Equal hashes mean an identical wire surface, in any port, so a caller holding a cached description for this hash can skip :func:describe_protocol. TYPE: str
deprecated	Whether callers should migrate off this protocol. TYPE: bool
deprecation_message	What to migrate to; empty unless deprecated. TYPE: str
features	Capability tokens the protocol announces. TYPE: tuple[str, ...]

ServiceDescription dataclass

ServiceDescription(
    protocol_name: str,
    request_version: str,
    describe_version: str,
    protocol_hash: str,
    server_id: str,
    methods: Mapping[str, MethodDescription],
    protocol_version: str = "",
)

Complete description of an RPC service from introspection.

ATTRIBUTE	DESCRIPTION
protocol_name	Name of the Protocol class. TYPE: str
request_version	Wire protocol version. TYPE: str
describe_version	Introspection format version. TYPE: str
protocol_hash	SHA-256 hex digest of the canonical describe payload. Stable across server processes that expose the same Protocol; changes when any wire-relevant detail changes. Use this as the schema-registry key when decoding archived access-log records. TYPE: str
server_id	Server instance identifier. TYPE: str
methods	Mapping of method name to MethodDescription. TYPE: Mapping[str, MethodDescription]
protocol_version	Application protocol surface version declared by the Protocol class (canonical semver MAJOR.MINOR.PATCH), or empty string when the Protocol opts out. Diagnostic — actual enforcement happens via the per-request vgi_rpc.protocol_version metadata key at the server's dispatch boundary. TYPE: str

__str__

__str__() -> str

Return a human-readable summary of the service.

Source code in vgi_rpc/introspect.py

def __str__(self) -> str:
    """Return a human-readable summary of the service."""
    lines: list[str] = [
        f"RPC Service: {self.protocol_name}",
        f"  server_id: {self.server_id}",
        f"  request_version: {self.request_version}",
        f"  describe_version: {self.describe_version}",
        f"  protocol_hash: {self.protocol_hash}",
    ]
    if self.protocol_version:
        lines.append(f"  protocol_version: {self.protocol_version}")
    lines.append("")
    for name, md in sorted(self.methods.items()):
        lines.append(f"  {name}({md.method_type.value})")
        if md.params_schema.names:
            lines.append(f"    params: {md.params_schema}")
        if md.has_return:
            lines.append(f"    returns: {md.result_schema}")
        lines.append("")
    return "\n".join(lines)

MethodDescription dataclass

MethodDescription(
    name: str,
    method_type: MethodType,
    has_return: bool,
    params_schema: Schema,
    result_schema: Schema,
    has_header: bool = False,
    header_schema: Schema | None = None,
    is_exchange: bool | None = None,
)

Description of a single RPC method from introspection.

The wire format carries only language-neutral data: name, method type, schemas, and stream flags. Rich Python-flavoured metadata (parameter type names, defaults, docstrings) lives in the Protocol source class — consumers needing that information import the Protocol directly rather than reconstructing it from the wire.

For STREAM methods, result_schema reflects the Protocol-level return type (always empty). The actual stream output schema is determined at runtime by the implementation and cannot be reported statically.

ATTRIBUTE	DESCRIPTION
name	Method name as it appears on the Protocol. TYPE: str
method_type	Whether this is UNARY or STREAM. TYPE: MethodType
has_return	True for unary methods that return a value. TYPE: bool
params_schema	Arrow schema for request parameters. TYPE: Schema
result_schema	Arrow schema for the response (unary) or empty (streams). TYPE: Schema
has_header	True for stream methods that declare a header type. TYPE: bool
header_schema	Arrow schema for the header, or None if no header. TYPE: Schema | None
is_exchange	For streams, True if exchange (bidi), False if producer, None if unknown. Always None for unary. TYPE: bool | None

Exceptions

ReflectionNotSupportedError

ReflectionNotSupportedError(
    error_type: str,
    error_message: str,
    remote_traceback: str,
    *,
    request_id: str = "",
    error_code: str = "",
    error_kind: str = "",
    error_details: list[dict[str, Any]] | None = None
)

Bases: RpcError

The server does not host vgi_rpc.Reflection.v1.

Raised by :func:list_protocols and :func:describe_protocol when the server answers the reflection call with "not hosted" rather than with a listing: a server built without enable_describe=True (the default), or one that predates reflection. Such a server still serves its own protocol, so this is a statement about discovery, not about the connection -- the connection remains usable.

A subclass of :class:~vgi_rpc.rpc.RpcError carrying the server's original error fields, so code that already catches RpcError keeps working; catch this class to branch on "cannot discover" specifically.

Source code in vgi_rpc/rpc/_common.py

def __init__(
    self,
    error_type: str,
    error_message: str,
    remote_traceback: str,
    *,
    request_id: str = "",
    error_code: str = "",
    error_kind: str = "",
    error_details: list[dict[str, Any]] | None = None,
) -> None:
    """Initialize with error details from the remote side.

    Args:
        error_type: The remote exception's class name.
        error_message: The remote message.
        remote_traceback: The remote traceback, or ``""`` when the server
            has tracebacks turned off.
        request_id: The request correlation ID, when known.
        error_code: The ``vgi_rpc.error_code`` value, or ``""``.
        error_kind: The ``vgi_rpc.error_kind`` value, or ``""``.
        error_details: The decoded ``vgi_rpc.error_details`` objects.

    """
    self.error_type = error_type
    self.error_message = error_message
    self.remote_traceback = remote_traceback
    self.request_id = request_id
    self.error_code = error_code
    self.error_kind = error_kind
    self.error_details: list[dict[str, Any]] = list(error_details) if error_details else []
    super().__init__(f"{error_type}: {error_message}")

from_rpc_error classmethod

from_rpc_error(
    error: RpcError,
) -> ReflectionNotSupportedError

Wrap the server's "not hosted" answer, keeping every field.

PARAMETER	DESCRIPTION
error	The error the reflection call raised. TYPE: RpcError

RETURNS	DESCRIPTION
ReflectionNotSupportedError	An instance carrying error's type, message, traceback, request
ReflectionNotSupportedError	id, code, kind and details.

Source code in vgi_rpc/introspect.py

@classmethod
def from_rpc_error(cls, error: RpcError) -> ReflectionNotSupportedError:
    """Wrap the server's "not hosted" answer, keeping every field.

    Args:
        error: The error the reflection call raised.

    Returns:
        An instance carrying *error*'s type, message, traceback, request
        id, code, kind and details.

    """
    return cls(
        error.error_type,
        error.error_message,
        error.remote_traceback,
        request_id=error.request_id,
        error_code=error.error_code,
        error_kind=error.error_kind,
        error_details=error.error_details,
    )

Constants

DESCRIBE_VERSION module-attribute

DESCRIBE_VERSION = '5'

Introspection format version.

Vestigial since v5: introspection is vgi_rpc.Reflection.v1 now, a protocol whose major version is part of its own name, so there is no separate format number to negotiate. Reported for the benefit of readers who still look for it, and it will not move again.

History

• v5: introspection became a protocol. __describe__, _DESCRIBE_SCHEMA, build_describe_batch and parse_describe_batch are gone; the payload is a generated schema and the hash is taken over the decoded description.
• v4: dropped Python-flavoured fields; introduced protocol_hash.
• v3: added param_docs_json.
• v2: added has_header, header_schema_ipc, is_exchange.