Inter-Service RPC

Understand Singularity's RPC system for local in-process calls and remote cross-deployment communication, including blacklist enforcement, Redis discovery, and the Microservice descriptor pattern.

Singularity provides a built-in RPC (Remote Procedure Call) system that lets services call each other's methods without tight coupling. Calls within the same process are dispatched directly in-memory. Calls to services running on separate deployments are routed transparently over HTTP, discovered via Redis.

The caller always identifies itself by name. The target service can define a blacklist to deny specific callers. From the caller's perspective, the syntax is identical whether the target is local or remote.

Core Concepts

RPCRegistry

Singleton that stores all local and remote service entries. Generates the RPC spec and handles Redis publish/discover.

RPCProxy

Returned by rpc("caller"). Attribute access resolves to a ServiceProxy for the named target service.

ServiceProxy

Dispatches method calls locally (in-process) or remotely (HTTP) with blacklist enforcement.

Microservice

Descriptor for declaring dependencies on services in entirely separate codebases, with per-repo URL and API key.

Making a Local RPC Call

To call another service's method, import rpc and use the chainable proxy syntax. The first argument is the caller's identity (used for blacklist checks). The next attribute is the target service name. The final attribute is the method name.

from singularity.rpc import rpc

class OrdersService:
  def __init__(self, acquire):
    self.acquire = acquire

  async def post(self, data: dict):
    """Create an order and charge the user via RPC."""
    payment = await rpc("orders").payments.charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {"order": "confirmed", "payment": payment}

Under the hood, this is what happens:

rpc("orders") creates an RPCProxy bound to the caller identity "orders".
.payments triggers __getattr__, which creates a ServiceProxy for the target "payments".
.charge(...) triggers __getattr__ on the ServiceProxy, which looks up "payments" in the registry.
The registry finds a LocalServiceEntry (in-process). The blacklist is checked -- "orders" is not blocked.
The actual charge method on the PaymentsService instance is called directly in memory.
The result is returned to the caller.

Local RPC calls have zero network overhead. They are plain Python method calls dispatched through the proxy layer. The only added cost is the blacklist check and registry lookup.

Exposing a Service for RPC

Set rpc_exposed = True on the service class. The Manager collects these during discovery and registers them in the RPCRegistry. Only methods that correspond to HTTP endpoints (core methods like get, post, etc., and methods declared in http_exposed) are exposed via RPC.

class PaymentsService:
  """Payment processing service."""
  rpc_exposed = True
  blacklist = ["public", "analytics"]

  def __init__(self, acquire):
    self.acquire = acquire

  async def charge(self, user_id: str, amount: float):
    """Charge a user's payment method."""
    # Process payment...
    return {"charged": amount, "user": user_id, "transaction_id": "txn_abc123"}

  async def refund(self, payment_id: str):
    """Refund a payment."""
    return {"refunded": True, "payment": payment_id}

  async def get(self):
    """List recent payments — also exposed as GET /api/v1/payments."""
    return {"payments": []}

All three methods (charge, refund, get) become callable via RPC by any service not in the blacklist.

The Blacklist

The blacklist attribute is a list of caller names that are denied access to this service via RPC. When a blacklisted caller attempts a call, RPCAccessDenied is raised immediately, before the method is invoked.

class PaymentsService:
  rpc_exposed = True
  blacklist = ["public", "analytics"]

  async def charge(self, user_id: str, amount: float):
    return {"charged": amount}

With this configuration:

rpc("orders").payments.charge(...) succeeds (orders is not blacklisted).
rpc("analytics").payments.charge(...) raises RPCAccessDenied.
rpc("public").payments.charge(...) raises RPCAccessDenied.

Blacklist Enforcement Flowchart

Every RPC call passes through this decision tree:

If a service is not found in the local registry or the cached remote registry, the proxy performs an on-demand Redis lookup (discover_single_service) before raising RPCServiceNotFound. This handles the case where a remote service registered after the initial discovery.

Remote RPC (Cross-Deployment)

When multiple Singularity instances run on different hosts, they discover each other through Redis. Each instance publishes its RPC spec to Redis on startup and refreshes it periodically via the heartbeat.

How Remote Calls Work

When the ServiceProxy finds the target in _remote (not _local), it constructs an HTTP request using httpx:

The endpoint URL comes from the remote service's spec (stored in Redis).
The HTTP method matches what the remote service declared (GET, POST, etc.).
An X-RPC-Caller header identifies the calling service.
Parameters are passed as query parameters.
The response is parsed as JSON and returned to the caller.

From the caller's perspective, the code is identical to a local call:

# This works whether payments is local or remote
result = await rpc("orders").payments.charge(user_id="abc", amount=9.99)

Redis Key Structure

Each service is published under a per-service key and a combined spec key:

Key	Content	TTL
`singularity:rpc:services:{name}`	Single service spec with `base_url`, `blacklist`, `methods`	60s
`singularity:rpc:spec`	Full spec for all services on this instance	60s

The Heartbeat

The Manager starts an asyncio background task that re-publishes the RPC spec to Redis every 30 seconds with a 60-second TTL. This ensures that:

If an instance crashes, its keys expire from Redis within 60 seconds.
Healthy instances always have fresh entries.
New instances can discover existing ones immediately on startup.

# From Manager._heartbeat_loop()
async def _heartbeat_loop(self) -> None:
    """Periodically refresh RPC registry keys in Redis."""
    while True:
        await asyncio.sleep(30)
        try:
            await registry.publish_to_redis(ttl=60)
        except Exception as e:
            logger.warning(f"RPC heartbeat failed: {e}")

To disable the heartbeat (useful during development or testing):

uv run python app.py --dev --no-heartbeat

This sets the SINGULARITY_NO_HEARTBEAT=1 environment variable, which the Manager checks at startup.

Disabling the heartbeat does not disable RPC itself. Local calls still work normally. It only stops the periodic Redis refresh, which means remote services may lose visibility of this instance after the TTL expires.

The RPC Spec Endpoint

Every Singularity instance exposes its full RPC spec at GET /_rpc/spec. This endpoint is excluded from the OpenAPI schema (include_in_schema=False) and is used for debugging and tooling.

curl http://localhost:8000/_rpc/spec | python -m json.tool

Example response:

{
  "rpc_version": "1.0",
  "base_url": "http://localhost:8000",
  "services": {
    "payments": {
      "blacklist": ["public", "analytics"],
      "methods": {
        "charge": {
          "endpoint": "http://localhost:8000/api/v1/payments",
          "http_method": "POST",
          "description": "Charge a user's payment method.",
          "parameters": {
            "user_id": { "type": "str", "required": true },
            "amount": { "type": "float", "required": true }
          },
          "returns": "Any"
        },
        "get": {
          "endpoint": "http://localhost:8000/api/v1/payments",
          "http_method": "GET",
          "description": "List recent payments.",
          "parameters": {},
          "returns": "Any"
        }
      }
    }
  }
}

The spec is generated by RPCRegistry.generate_spec(), which introspects each registered service's methods, extracts their signatures, and produces an OpenAPI-like document.

The Microservice Descriptor

For calling services that live in entirely separate codebases (not just separate instances of the same codebase), use the Microservice descriptor. This provides a clean, class-based interface with per-repo configuration.

from singularity.microservices.microservice import Microservice
from singularity.config import settings

class BillingMS(Microservice):
  host_url = settings.billing_host_url  # e.g., "http://billing-service:8001"
  api_key = settings.billing_api_key     # sent as X-API-Key header
  timeout = 10                            # request timeout in seconds

Use it as a class attribute on your service:

class InvoiceService:
  billing = BillingMS()

  def __init__(self, acquire):
    self.acquire = acquire

  async def post(self, data: dict):
    """Generate an invoice via the billing microservice."""
    # Calls POST http://billing-service:8001/api/invoice
    result = await self.billing.invoice.post(order_id=data["order_id"])
    return {"invoice": result}

How the Descriptor Works

Microservice is a Python descriptor. When accessed on an instance (e.g., self.billing), its __get__ method returns a MicroserviceProxy bound to the caller's identity:

class Microservice:
  host_url: str = ""
  api_key: str = ""
  timeout: float = 30.0
  service_name: str = ""

  def __get__(self, instance, owner):
    if instance is None:
      return self
    caller = getattr(instance, "_service_name", "unknown")
    if self.service_name:
      # Single-service shortcut: skip the repo proxy
      return MicroserviceServiceProxy(
        service_name=self.service_name, host_url=self.host_url,
        api_key=self.api_key, timeout=self.timeout, caller=caller,
      )
    return MicroserviceProxy(
      host_url=self.host_url, api_key=self.api_key,
      timeout=self.timeout, caller=caller,
    )

Multi-Service vs Single-Service Mode

When service_name is not set, accessing the descriptor returns a MicroserviceProxy. You then specify the target service as the next attribute:

class BillingMS(Microservice):
  host_url = "http://billing:8001"

class OrdersService:
  billing = BillingMS()

  async def checkout(self):
    # billing.invoice -> MicroserviceServiceProxy for "invoice"
    # .post(...) -> POST http://billing:8001/api/invoice
    result = await self.billing.invoice.post(order_id="123")

    # billing.subscription -> MicroserviceServiceProxy for "subscription"
    # .get() -> GET http://billing:8001/api/subscription
    status = await self.billing.subscription.get()

When service_name is set, the descriptor returns a MicroserviceServiceProxy directly, skipping the intermediate repo proxy:

class TestMS(Microservice):
  host_url = "http://test-service:8002"
  service_name = "test"

class ExampleService:
  test = TestMS()

  async def get(self):
    # self.test.get() -> GET http://test-service:8002/api/test
    result = await self.test.get()
    return result

URL Construction Convention

Method names are mapped to HTTP verbs and endpoints using the same convention as the Manager's route registration:

Method call	HTTP Method	Endpoint
`.get()`	`GET`	`{host_url}/api/{service}`
`.post(...)`	`POST`	`{host_url}/api/{service}`
`.get_status(...)`	`GET`	`{host_url}/api/{service}/status`
`.post_order_payment(...)`	`POST`	`{host_url}/api/{service}/order_payment`

Core HTTP verbs (get, post, put, delete, patch) map to the base service endpoint. Prefixed methods (get_xxx, post_xxx) map to sub-paths.

Declarative RPC Dependencies with `rpc.remote()`

For services that always call the same target, you can use the rpc.remote() descriptor instead of calling rpc() at runtime:

from singularity.rpc import rpc

class OrdersService:
  # Declare dependency at class level
  payments = rpc.remote("payments")

  async def post(self, data: dict):
    # Uses the class attribute — caller identity is auto-injected
    result = await self.payments.charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {"order": "confirmed", "payment": result}

The RemoteServiceDescriptor is a Python descriptor that, when accessed on an instance, reads the _service_name attribute (injected by the Manager) and creates a bound ServiceProxy. This is equivalent to calling rpc("orders")["payments"] but declared once at the class level.

Error Handling

The RPC module defines a hierarchy of exceptions that cover every failure mode:

RPCErrorBase class for all RPC errors.

RPCServiceNotFoundTarget service is not in the local registry, not in the remote cache, and not discoverable from Redis.
RPCMethodNotFoundTarget service exists but does not expose the requested method.
RPCAccessDeniedThe caller is in the target service's blacklist.
RPCServiceUnavailableRemote service was found but the HTTP connection failed (host unreachable, timeout, etc.).

from singularity.rpc import (
    rpc,
    RPCServiceNotFound,
    RPCAccessDenied,
    RPCMethodNotFound,
    RPCServiceUnavailable,
)

class OrdersService:
  async def post(self, data: dict):
    try:
      result = await rpc("orders").payments.charge(
        user_id=data["user_id"],
        amount=data["total"],
      )
      return {"order": "confirmed", "payment": result}

    except RPCAccessDenied:
      return {"error": "Orders service is not allowed to access payments"}

    except RPCServiceNotFound:
      return {"error": "Payments service is not registered or unreachable"}

    except RPCMethodNotFound:
      return {"error": "The charge method does not exist on payments"}

    except RPCServiceUnavailable as e:
      return {"error": f"Payments service is down: {e}"}

For remote calls, httpx.HTTPStatusError can also be raised if the remote service returns a non-2xx status code. This is re-raised directly and is not wrapped in an RPCError subclass. Handle it separately if you need to inspect the HTTP response.

Startup Sequence

Understanding when RPC initialization happens helps with debugging:

Module load: Manager.register_services() scans for services and collects those with rpc_exposed = True into _rpc_pending.
RPC init: _init_rpc() iterates over _rpc_pending, calling registry.register(name, instance, path_segments) for each. The spec is generated and logged.
Lifespan startup: manager.startup() is called by FastAPI's lifespan:
- registry.publish_to_redis(ttl=60) publishes each service's spec as a Redis key.
- registry.discover_remote() scans Redis for keys matching singularity:rpc:services:* and populates _remote.
- If the heartbeat is enabled and RPC services exist, _heartbeat_loop is started as an asyncio task.
Serving: The app accepts requests. RPC calls work for both local and discovered remote services.
Lifespan shutdown: manager.shutdown() cancels the heartbeat task and calls registry.remove_from_redis() to clean up Redis keys.

Full Working Example

Here is a complete example with two services that communicate via RPC:

# services/v1/payments/service.py
"""Payment processing service."""

from singularity.core.acquire import Acquire


class PaymentsService:
  """Handles charging and refunding payments."""

  rpc_exposed = True
  blacklist = ["public"]

  def __init__(self, acquire: Acquire):
    self.acquire = acquire
    self.db = acquire.db_session

  async def get(self):
    """List recent payments."""
    return {"payments": []}

  async def charge(self, user_id: str, amount: float):
    """Charge a user's payment method."""
    async with self.db() as session:
      # Record the payment
      payment = Payment(user_id=user_id, amount=amount, status="charged")
      session.add(payment)
      await session.commit()
      return {
        "charged": amount,
        "user": user_id,
        "transaction_id": payment.id,
      }

  async def refund(self, payment_id: str):
    """Refund a payment."""
    return {"refunded": True, "payment": payment_id}

# services/v1/orders/service.py
"""Order management service."""

from singularity.core.acquire import Acquire
from singularity.rpc import rpc


class OrdersService:
  """Creates orders and coordinates payment via RPC."""

  rpc_exposed = True

  def __init__(self, acquire: Acquire):
    self.acquire = acquire

  async def get(self):
    """List orders."""
    return {"orders": []}

  async def post(self, data: dict):
    """Create an order and charge the user."""
    payment = await rpc("orders").payments.charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {
      "order_id": "ord_new",
      "status": "confirmed",
      "payment": payment,
    }

Scaffold both with:

singularity generate service v1/payments -d "Payment processing" --template crud --rpc --db
singularity generate service v1/orders -d "Order management" --template crud --rpc

RPCRegistry

RPCProxy

ServiceProxy

Microservice

On this page