orchestrata_ultimate.gateway.method_spec

Reference › orchestrata_ultimate › gateway › method_spec

MethodSpec is one row of the gateway's declarative method table (SPEC-014 REQ-014-203, ADR-0099). It is a frozen dataclass holding the JSON-RPC method name, the dotted path <module>.<Class>.<method> of the handler coroutine, a mutates flag, an expose flag, an optional params schema and a one-line description. mutates defaults to true, so a method counts as mutating unless its handler was read and confirmed read-only. expose is false for streaming methods, legacy aliases and auth.authenticate, which stay RPC-only and never become MCP tools. The properties owner and attribute split the handler path; input_schema returns the params schema or a bare object schema; resolve_owner_class imports the handler class. Used by method_table, rpc, mcp_tool_spec and mcp_transport.

At a glance

Class diagram

Dependency diagram

API

Classes

MethodSpec (dataclass)

Decorators: traces(spec='014', story='US-013', reqs=['REQ-014-203']), dataclass(frozen=True) · Line 26

One gateway JSON-RPC method.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/method_spec.py

"""MethodSpec -- one row of the gateway's declarative method table.

ADR-0099 rule 2: ``METHODS`` (``gateway/method_table.py``) is the single
source both transports read -- ``RPCHandler`` builds its dispatch dict from
it and ``McpTransport`` builds its tool list from it. A spec holds only data
(a handler *dotted path*, never a bound method), so the table imports
without constructing any handler or infrastructure.

Author: Mohamed
"""

# Spec: 014 - WebSocket Gateway

from __future__ import annotations

import importlib
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from orchestrata_ultimate.core.base.traces import traces


@traces(spec="014", story="US-013", reqs=["REQ-014-203"])
@dataclass(frozen=True)
class MethodSpec:
    """One gateway JSON-RPC method.

    Attributes:
        name: JSON-RPC method name (``"volk.list"``).
        handler: Dotted path ``<module>.<Class>.<method>`` of the handler
            coroutine; resolved against a handler *instance* by the caller.
        mutates: ``True`` unless the handler was read and confirmed
            read-only (ADR-0099 rule 3: unmarked methods fail closed).
        expose: ``False`` for streaming methods, legacy aliases and
            ``auth.authenticate`` -- RPC-only, never an MCP tool.
        params_schema: Optional JSON schema for the params object; with none
            the tool's schema is ``{"type": "object"}`` and the gateway's own
            validation is the contract.
        description: One line shown as the MCP tool description.
    """

    name: str
    handler: str
    mutates: bool = True
    expose: bool = True
    params_schema: Mapping[str, Any] | None = None
    description: str = ""

    @property
    def owner(self) -> str:
        """Dotted path of the handler class (``<module>.<Class>``)."""
        return self.handler.rpartition(".")[0]

    @property
    def attribute(self) -> str:
        """Name of the handler method on the owner class."""
        return self.handler.rpartition(".")[2]

    def input_schema(self) -> dict[str, Any]:
        """The JSON schema for this method's params object."""
        if self.params_schema:
            return dict(self.params_schema)
        return {"type": "object"}

    def resolve_owner_class(self) -> type:
        """Import and return the owner class named by ``handler``."""
        module_path, _, class_name = self.owner.rpartition(".")
        owner_class: type = getattr(importlib.import_module(module_path), class_name)
        return owner_class