Skip to content

Nexus samples #174

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 46 commits into from
Jul 11, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
46 commits
Select commit Hold shift + click to select a range
266dd3c
Nexus samples
dandavison Feb 20, 2025
dac5950
Install SDKs from github
dandavison Jun 9, 2025
306bb34
Change to endpoint_description.md
dandavison Jun 9, 2025
8c6f43b
Revert "Install SDKs from github"
dandavison Jun 9, 2025
ba24332
Get rid of type hint
dandavison Jun 9, 2025
4b06bb8
Remove unnecessary re-imports
dandavison Jun 9, 2025
d71c0e4
Fix test
dandavison Jun 9, 2025
fd3dc44
Update to use Temporal operation contexts
dandavison Jun 12, 2025
bb5114f
Emphasize that StartOperationContext is from Temporal
dandavison Jun 12, 2025
99e518e
s/target/handler/
dandavison Jun 19, 2025
2d7da19
Respond to upstream rename
dandavison Jun 19, 2025
df61c94
Respond to upstream: NexusStartWorkflowRequest
dandavison Jun 22, 2025
306418d
Respond to upstream: tctc.start_workflow -> WorkflowOperationToken
dandavison Jun 22, 2025
ad54378
Respond to upstream: use factories instead of decorators
dandavison Jun 23, 2025
c986a84
uv.lock
dandavison Jun 23, 2025
f8643e7
Respond to upstream: temporal_operation_context
dandavison Jun 24, 2025
b7a361f
Respond to upstream: top-level start_workflow function
dandavison Jun 24, 2025
df4794a
Respond to upstream: from_callable, inherit from abstract base class
dandavison Jun 25, 2025
3de98fc
Cleanup
dandavison Jun 25, 2025
66016e7
Respond to upstream
dandavison Jun 26, 2025
65f76a0
Delete operations-as-classes sample
dandavison Jun 26, 2025
1ed070c
Respond to upstream: rename decorators
dandavison Jun 26, 2025
cd7e16d
Rename service class
dandavison Jun 26, 2025
4c10816
Cleanup
dandavison Jun 26, 2025
fe3f23e
RTU: dependencies
dandavison Jun 26, 2025
4513de4
Pass through imports
dandavison Jun 26, 2025
25da5a0
Fix directory paths in README
dandavison Jun 26, 2025
2565763
Make namespace/task queue/enspoint names sample-specific
dandavison Jun 26, 2025
6667d5e
RTU: ctx.start_workflow()
dandavison Jun 26, 2025
1ae914a
Delete no-service-definition sample
dandavison Jun 26, 2025
96f7d1d
Reorganize
dandavison Jun 26, 2025
2da62a0
Enable sandbox in Nexus sample
dandavison Jun 26, 2025
d4977c6
uv.lock
dandavison Jun 27, 2025
4b4e520
uv.lock
dandavison Jun 30, 2025
acdf591
RTU: Nexus client
dandavison Jul 7, 2025
080b804
Install SDKs from github
dandavison Jul 7, 2025
c41fb80
Don't demonstrate passing in a db client
dandavison Jul 8, 2025
a38d93f
Add nexus-rpc version constraint
dandavison Jul 8, 2025
c90e714
Use SDK from github
dandavison Jul 8, 2025
d5a5610
Revert "Use SDK from github"
dandavison Jul 8, 2025
97507bd
dependencies
dandavison Jul 9, 2025
6444d39
Run commands from repo root dir
dandavison Jul 9, 2025
6d4a0c2
Edits, imports
dandavison Jul 9, 2025
885920a
Skip nexus tests under java test server
dandavison Jul 10, 2025
a134d6d
1.14.1
dandavison Jul 10, 2025
0c3fc1d
Skip on 3.9
dandavison Jul 11, 2025
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions hello_nexus/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
This sample shows how to define a Nexus service, implement the operation handlers, and
call the operations from a workflow.

### Sample directory structure

- [service.py](./service.py) - shared Nexus service definition
- [caller](./caller) - a caller workflow that executes Nexus operations, together with a worker and starter code
- [handler](./handler) - Nexus operation handlers, together with a workflow used by one of the Nexus operations, and a worker that polls for both workflow and Nexus tasks.


### Instructions

Start a Temporal server. (See the main samples repo [README](../README.md)).

Run the following:

```
temporal operator namespace create --namespace hello-nexus-basic-handler-namespace
temporal operator namespace create --namespace hello-nexus-basic-caller-namespace

temporal operator nexus endpoint create \
--name hello-nexus-basic-nexus-endpoint \
--target-namespace hello-nexus-basic-handler-namespace \
--target-task-queue my-handler-task-queue \
--description-file hello_nexus/endpoint_description.md
```

In one terminal, run the Temporal worker in the handler namespace:
```
uv run hello_nexus/handler/worker.py
```

In another terminal, run the Temporal worker in the caller namespace and start the caller
workflow:
```
uv run hello_nexus/caller/app.py
```
Empty file added hello_nexus/__init__.py
Empty file.
Empty file.
43 changes: 43 additions & 0 deletions hello_nexus/caller/app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
import asyncio
import uuid
from typing import Optional

from temporalio.client import Client
from temporalio.worker import Worker

from hello_nexus.caller.workflows import CallerWorkflow
from hello_nexus.service import MyOutput

NAMESPACE = "hello-nexus-basic-caller-namespace"
TASK_QUEUE = "hello-nexus-basic-caller-task-queue"


async def execute_caller_workflow(
client: Optional[Client] = None,
) -> tuple[MyOutput, MyOutput]:
client = client or await Client.connect(
"localhost:7233",
namespace=NAMESPACE,
)

async with Worker(
client,
task_queue=TASK_QUEUE,
workflows=[CallerWorkflow],
):
return await client.execute_workflow(
CallerWorkflow.run,
arg="world",
id=str(uuid.uuid4()),
task_queue=TASK_QUEUE,
)


if __name__ == "__main__":
loop = asyncio.new_event_loop()
try:
results = loop.run_until_complete(execute_caller_workflow())
for output in results:
print(output.message)
except KeyboardInterrupt:
loop.run_until_complete(loop.shutdown_asyncgens())
35 changes: 35 additions & 0 deletions hello_nexus/caller/workflows.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
from hello_nexus.service import MyInput, MyNexusService, MyOutput

NEXUS_ENDPOINT = "hello-nexus-basic-nexus-endpoint"


# This is a workflow that calls two nexus operations.
@workflow.defn
class CallerWorkflow:
# An __init__ method is always optional on a workflow class. Here we use it to set the
# nexus client, but that could alternatively be done in the run method.
def __init__(self):
self.nexus_client = workflow.create_nexus_client(
service=MyNexusService,
endpoint=NEXUS_ENDPOINT,
)

# The workflow run method invokes two nexus operations.
@workflow.run
async def run(self, name: str) -> tuple[MyOutput, MyOutput]:
# Start the nexus operation and wait for the result in one go, using execute_operation.
wf_result = await self.nexus_client.execute_operation(
MyNexusService.my_workflow_run_operation,
MyInput(name),
)
# Alternatively, you can use start_operation to obtain the operation handle and
# then `await` the handle to obtain the result.
sync_operation_handle = await self.nexus_client.start_operation(
MyNexusService.my_sync_operation,
MyInput(name),
)
sync_result = await sync_operation_handle
return sync_result, wf_result
3 changes: 3 additions & 0 deletions hello_nexus/endpoint_description.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
## Service: [MyNexusService](https://github.com/temporalio/samples-python/blob/main/hello_nexus/basic/service.py)
- operation: `my_sync_operation`
- operation: `my_workflow_run_operation`
Empty file.
49 changes: 49 additions & 0 deletions hello_nexus/handler/service_handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
"""
This file demonstrates how to implement a Nexus service.
"""

from __future__ import annotations

import uuid

import nexusrpc
from temporalio import nexus

from hello_nexus.handler.workflows import WorkflowStartedByNexusOperation
from hello_nexus.service import MyInput, MyNexusService, MyOutput


@nexusrpc.handler.service_handler(service=MyNexusService)
class MyNexusServiceHandler:
# You can create an __init__ method accepting what is needed by your operation
# handlers to handle requests. You typically instantiate your service handler class
# when starting your worker. See hello_nexus/basic/handler/worker.py.

# This is a nexus operation that is backed by a Temporal workflow. The start method
# starts a workflow, and returns a nexus operation token. Meanwhile, the workflow
# executes in the background; Temporal server takes care of delivering the eventual
# workflow result (success or failure) to the calling workflow.
#
# The token will be used by the caller if it subsequently wants to cancel the Nexus
# operation.
@nexus.workflow_run_operation
async def my_workflow_run_operation(
self, ctx: nexus.WorkflowRunOperationContext, input: MyInput
) -> nexus.WorkflowHandle[MyOutput]:
return await ctx.start_workflow(
WorkflowStartedByNexusOperation.run,
input,
id=str(uuid.uuid4()),
)

# This is a Nexus operation that responds synchronously to all requests. That means
# that unlike the workflow run operation above, in this case the `start` method
# returns the final operation result.
#
# Sync operations are free to make arbitrary network calls, or perform CPU-bound
# computations. Total execution duration must not exceed 10s.
@nexusrpc.handler.sync_operation
async def my_sync_operation(
self, ctx: nexusrpc.handler.StartOperationContext, input: MyInput
) -> MyOutput:
return MyOutput(message=f"Hello {input.name} from sync operation!")
46 changes: 46 additions & 0 deletions hello_nexus/handler/worker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
import asyncio
import logging
from typing import Optional

from temporalio.client import Client
from temporalio.worker import Worker

from hello_nexus.handler.service_handler import MyNexusServiceHandler
from hello_nexus.handler.workflows import WorkflowStartedByNexusOperation

interrupt_event = asyncio.Event()

NAMESPACE = "hello-nexus-basic-handler-namespace"
TASK_QUEUE = "my-handler-task-queue"


async def main(client: Optional[Client] = None):
logging.basicConfig(level=logging.INFO)

client = client or await Client.connect(
"localhost:7233",
namespace=NAMESPACE,
)

# Start the worker, passing the Nexus service handler instance, in addition to the
# workflow classes that are started by your nexus operations, and any activities
# needed. This Worker will poll for both workflow tasks and Nexus tasks (this example
# doesn't use any activities).
async with Worker(
client,
task_queue=TASK_QUEUE,
workflows=[WorkflowStartedByNexusOperation],
nexus_service_handlers=[MyNexusServiceHandler()],
):
logging.info("Worker started, ctrl+c to exit")
await interrupt_event.wait()
logging.info("Shutting down")


if __name__ == "__main__":
loop = asyncio.new_event_loop()
try:
loop.run_until_complete(main())
except KeyboardInterrupt:
interrupt_event.set()
loop.run_until_complete(loop.shutdown_asyncgens())
12 changes: 12 additions & 0 deletions hello_nexus/handler/workflows.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we make this not required? Seems like everything in nexusrpc should be safe to import (except for the handler submodule)

from hello_nexus.service import MyInput, MyOutput


# This is the workflow that is started by the `my_workflow_run_operation` nexus operation.
@workflow.defn
class WorkflowStartedByNexusOperation:
@workflow.run
async def run(self, input: MyInput) -> MyOutput:
return MyOutput(message=f"Hello {input.name} from workflow run operation!")
33 changes: 33 additions & 0 deletions hello_nexus/service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
"""
This is a Nexus service definition.

A service definition defines a Nexus service as a named collection of operations, each
with input and output types. It does not implement operation handling: see the service
handler and operation handlers in hello_nexus.handler.nexus_service for that.

A Nexus service definition is used by Nexus callers (e.g. a Temporal workflow) to create
type-safe clients, and it is used by Nexus handlers to validate that they implement
correctly-named operation handlers with the correct input and output types.

The service defined in this file features two operations: echo and hello.
"""

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]
my_workflow_run_operation: nexusrpc.Operation[MyInput, MyOutput]
2 changes: 1 addition & 1 deletion open_telemetry/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.resources import SERVICE_NAME, Resource # type: ignore
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from temporalio import activity, workflow
Expand Down
12 changes: 10 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }]
requires-python = "~=3.9"
readme = "README.md"
license = "MIT"
dependencies = ["temporalio>=1.14.0,<2"]
dependencies = ["temporalio>=1.14.1,<2"]

[project.urls]
Homepage = "https://github.com/temporalio/samples-python"
Expand All @@ -22,7 +22,9 @@ dev = [
"pytest>=7.1.2,<8",
"pytest-asyncio>=0.18.3,<0.19",
"frozenlist>=1.4.0,<2",
"pyright>=1.1.394",
"types-pyyaml>=6.0.12.20241230,<7",
"pytest-pretty>=1.3.0",
]
bedrock = ["boto3>=1.34.92,<2"]
dsl = [
Expand All @@ -44,9 +46,12 @@ langchain = [
"tqdm>=4.62.0,<5",
"uvicorn[standard]>=0.24.0.post1,<0.25",
]
nexus = [
"nexus-rpc>=1.1.0,<2",
]
open-telemetry = [
"temporalio[opentelemetry]",
"opentelemetry-exporter-otlp-proto-grpc==1.18.0",
"opentelemetry-exporter-otlp-proto-grpc",
]
openai-agents = [
"openai-agents >= 0.0.19",
Expand All @@ -73,6 +78,7 @@ default-groups = [
"encryption",
"gevent",
"langchain",
"nexus",
"open-telemetry",
"pydantic-converter",
"sentry",
Expand All @@ -98,6 +104,7 @@ packages = [
"hello",
"langchain",
"message_passing",
"nexus",
"open_telemetry",
"patching",
"polling",
Expand Down Expand Up @@ -150,3 +157,4 @@ ignore_errors = true
[[tool.mypy.overrides]]
module = "opentelemetry.*"
ignore_errors = true

51 changes: 51 additions & 0 deletions tests/hello_nexus/hello_nexus_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
import asyncio
import sys

import pytest
from temporalio.client import Client
from temporalio.testing import WorkflowEnvironment

import hello_nexus.caller.app
import hello_nexus.caller.workflows
import hello_nexus.handler.worker
from tests.hello_nexus.helpers import create_nexus_endpoint, delete_nexus_endpoint


async def test_nexus_service_basic(client: Client, env: WorkflowEnvironment):
if env.supports_time_skipping:
pytest.skip("Nexus tests don't work under the Java test server")

if sys.version_info[:2] < (3, 10):
pytest.skip("Sample is written for Python >= 3.10")

create_response = await create_nexus_endpoint(
name=hello_nexus.caller.workflows.NEXUS_ENDPOINT,
task_queue=hello_nexus.handler.worker.TASK_QUEUE,
client=client,
)
try:
handler_worker_task = asyncio.create_task(
hello_nexus.handler.worker.main(
client,
)
)
await asyncio.sleep(1)
results = await hello_nexus.caller.app.execute_caller_workflow(
client,
)
hello_nexus.handler.worker.interrupt_event.set()
await handler_worker_task
hello_nexus.handler.worker.interrupt_event.clear()
print("\n\n")
print([r.message for r in results])
print("\n\n")
assert [r.message for r in results] == [
"Hello world from sync operation!",
"Hello world from workflow run operation!",
]
finally:
await delete_nexus_endpoint(
id=create_response.endpoint.id,
version=create_response.endpoint.version,
client=client,
)
Loading
Loading