Nexus RPC Python SDK

1.4.0 · active · verified Sun Apr 05

nexus-rpc is a Python SDK for building handlers that implement the Nexus synchronous RPC protocol. Nexus is designed for arbitrary-duration operations, allowing handlers to respond synchronously or return a token for asynchronous completion. As of version 1.4.0, the SDK is in an experimental release stage, and backwards-incompatible changes are expected prior to a stable release.

Warnings

breaking The nexus-rpc SDK is currently at an experimental release stage. Backwards-incompatible changes are anticipated until a stable release is announced.
Fix: Monitor official releases and changelogs for breaking changes. Pin exact versions in your `requirements.txt` to prevent unexpected updates.
gotcha The PyPI package name 'nexus-rpc' was previously used by an unrelated project. Ensure you are using the correct SDK for the Nexus synchronous RPC protocol from the `nexus-rpc` GitHub organization.
Fix: Verify the project's origin and documentation (github.com/nexus-rpc/sdk-python) to confirm you are using the intended library.
gotcha The SDK for the Nexus RPC protocol only provides components for implementing handlers. It does not include reference implementations for Nexus servers or clients.
Fix: You will need to implement or integrate a Nexus server and client separately to utilize the defined services and handlers. If integrating with Temporal, refer to their Nexus documentation for server-side setup.

Install

pip install nexus-rpc Install with pip

Imports

nexusrpc
```
import nexusrpc
```
service
```
from nexusrpc import service
```
Operation
```
from nexusrpc import Operation
```
handler
```
from nexusrpc import handler
```

Quickstart

This quickstart demonstrates how to define a Nexus service and implement a synchronous operation handler. It uses dataclasses for input and output types, decorates a class as a Nexus service, and then implements a service handler with a decorated synchronous operation.

from dataclasses import dataclass
import nexusrpc

@dataclass
class MyInput:
    name: str

@dataclass
class MyOutput:
    message: str

@nexusrpc.service
class MyNexusService:
    my_sync_operation: nexusrpc.Operation[MyInput, MyOutput]

@nexusrpc.handler.service_handler(service=MyNexusService)
class MyNexusServiceHandler:
    @nexusrpc.handler.sync_operation
    async def my_sync_operation(
        self, ctx: nexusrpc.handler.StartOperationContext, input: MyInput
    ) -> MyOutput:
        return MyOutput(message=f"Hello {input.name}!")

# In a real application, you would instantiate MyNexusServiceHandler
# and integrate it with your server/worker to process Nexus requests.
# For example:
# handler_instance = MyNexusServiceHandler()

view raw JSON →