Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ Some examples require extra dependencies. See each sample's directory for specif
This contains two samples, one sending messages to an existing workflow and a second that creates a workflow through Nexus
and sends messages to it.
* [nexus_multiple_args](nexus_multiple_args) - Map a Nexus operation to a handler workflow that takes multiple arguments.
* [nexus_standalone_activity](nexus_standalone_activity) - Back a Nexus operation with a standalone Activity.
* [nexus_standalone_operations](nexus_standalone_operations) - Execute Nexus operations directly from client code,
without wrapping them in a workflow.
* [open_telemetry](open_telemetry) - Trace workflows with OpenTelemetry.
Expand Down
59 changes: 59 additions & 0 deletions nexus_standalone_activity/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
# Nexus operation backed by a standalone Activity

This sample shows how to implement a `TemporalOperationHandler` that starts a
standalone Activity as the backing execution for a Nexus operation. When the Activity
finishes, Temporal delivers its result to the Nexus caller. The default handler
cancellation implementation also forwards Nexus cancellation to the Activity.

The APIs used by this sample are experimental and may change incompatibly.

### Sample structure

- [service.py](./service.py) defines the Nexus service shared by caller and handler.
- [activity.py](./activity.py) defines the standalone Activity.
- [handler.py](./handler.py) implements `TemporalOperationHandler.start_operation`.
- [worker.py](./worker.py) hosts the Nexus handler and Activity.
- [starter.py](./starter.py) executes the Nexus operation from client code.

## Run locally

This sample requires the [Temporal dev server build that supports standalone Nexus operations](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support) and Activity
callbacks enabled.

1. Start the server with caller and handler namespaces:

```bash
./temporal server start-dev \
--dynamic-config-value activity.enableCallbacks=true \
--namespace nexus-standalone-activity-caller \
--namespace nexus-standalone-activity-handler
```

2. Create an endpoint targeting the handler namespace and task queue:

```bash
./temporal operator nexus endpoint create \
--name nexus-standalone-activity-endpoint \
--target-namespace nexus-standalone-activity-handler \
--target-task-queue nexus-standalone-activity-handler
```

3. Start the handler Worker:

```bash
TEMPORAL_NAMESPACE=nexus-standalone-activity-handler \
uv run nexus_standalone_activity/worker.py
```

4. Execute the operation from the caller namespace:

```bash
TEMPORAL_NAMESPACE=nexus-standalone-activity-caller \
uv run nexus_standalone_activity/starter.py
```

Expected output:

```text
Hello, World!
```
1 change: 1 addition & 0 deletions nexus_standalone_activity/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Nexus operation backed by a standalone Activity sample."""
10 changes: 10 additions & 0 deletions nexus_standalone_activity/activity.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
"""Activity used as the backing execution for the Nexus operation."""

from temporalio import activity

from nexus_standalone_activity.service import GreetingInput, GreetingOutput


@activity.defn
async def create_greeting(input: GreetingInput) -> GreetingOutput:
return GreetingOutput(message=f"Hello, {input.name}!")
36 changes: 36 additions & 0 deletions nexus_standalone_activity/handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""Temporal operation handler that starts a standalone Activity."""

from datetime import timedelta

import nexusrpc.handler
from temporalio import nexus

from nexus_standalone_activity.activity import create_greeting
from nexus_standalone_activity.service import (
GreetingInput,
GreetingOutput,
GreetingService,
)


def get_activity_id(input: GreetingInput) -> str:
return f"greeting-{input.name}"


@nexusrpc.handler.service_handler(service=GreetingService)
class GreetingServiceHandler:
@nexus.temporal_operation
async def greet(
self,
_ctx: nexus.TemporalStartOperationContext,
client: nexus.TemporalNexusClient,
input: GreetingInput,
) -> nexus.TemporalOperationResult[GreetingOutput]:
# The standalone Activity becomes the asynchronous backing execution for
# this Nexus operation. Omitting task_queue uses the Nexus Worker's queue.
return await client.start_activity(
create_greeting,
input,
id=get_activity_id(input),
start_to_close_timeout=timedelta(seconds=10),
)
20 changes: 20 additions & 0 deletions nexus_standalone_activity/service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
"""Nexus service definition shared by the caller and handler."""

from dataclasses import dataclass

import nexusrpc


@dataclass
class GreetingInput:
name: str


@dataclass
class GreetingOutput:
message: str


@nexusrpc.service
class GreetingService:
greet: nexusrpc.Operation[GreetingInput, GreetingOutput]
34 changes: 34 additions & 0 deletions nexus_standalone_activity/starter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
"""Client that executes the activity-backed Nexus operation."""

import asyncio
import uuid
from datetime import timedelta

from temporalio.client import Client
from temporalio.envconfig import ClientConfig

from nexus_standalone_activity.service import GreetingInput, GreetingService

ENDPOINT_NAME = "nexus-standalone-activity-endpoint"


async def main() -> None:
config = ClientConfig.load_client_connect_config()
_ = config.setdefault("target_host", "localhost:7233")
client = await Client.connect(**config)

nexus_client = client.create_nexus_client(
service=GreetingService,
endpoint=ENDPOINT_NAME,
)
result = await nexus_client.execute_operation(
GreetingService.greet,
GreetingInput(name="World"),
id=f"greeting-{uuid.uuid4()}",
schedule_to_close_timeout=timedelta(seconds=10),
)
print(result.message)


if __name__ == "__main__":
asyncio.run(main())
41 changes: 41 additions & 0 deletions nexus_standalone_activity/worker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
"""Worker hosting the Nexus handler and its standalone Activity."""

import asyncio
import logging

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

from nexus_standalone_activity.activity import create_greeting
from nexus_standalone_activity.handler import GreetingServiceHandler

TASK_QUEUE = "nexus-standalone-activity-handler"

interrupt_event = asyncio.Event()


async def main() -> None:
logging.basicConfig(level=logging.INFO)

config = ClientConfig.load_client_connect_config()
_ = config.setdefault("target_host", "localhost:7233")
client = await Client.connect(**config)

async with Worker(
client,
task_queue=TASK_QUEUE,
activities=[create_greeting],
nexus_service_handlers=[GreetingServiceHandler()],
):
logging.info("Worker started, ctrl+c to exit")
_ = await interrupt_event.wait()


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())
11 changes: 5 additions & 6 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,10 @@ authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }]
requires-python = ">=3.10"
readme = "README.md"
license = "MIT"
dependencies = ["temporalio>=1.31.0,<2", "protobuf>=5.29.6,<6"]
dependencies = [
"temporalio @ git+https://github.com/temporalio/sdk-python.git@main",
"protobuf>=5.29.6,<6",
]

[project.urls]
Homepage = "https://github.com/temporalio/samples-python"
Expand Down Expand Up @@ -38,11 +41,7 @@ external-storage = [
]
external-storage-redis = ["redis>=5.0.0,<8"]
gevent = ["gevent>=25.4.2 ; python_version >= '3.8'"]
google-adk = ["temporalio[google-adk] >= 1.31.0", "google-adk>=2.2.0,<3"]
google-genai = [
"mcp>=1.0.0",
"temporalio[google-genai,pydantic]>=1.31.0",
]
google-adk = ["temporalio[google-adk] >= 1.30.0", "google-adk>=2.2.0,<3"]
langfuse-tracing = [
"openai>=1.4.0",
"temporalio[opentelemetry]>=1.30.0,<2",
Expand Down
4 changes: 4 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,10 @@ async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]:
env_type = request.config.getoption("--workflow-environment")
if env_type == "local":
env = await WorkflowEnvironment.start_local(
dev_server_extra_args=[
"--dynamic-config-value",
"activity.enableCallbacks=true",
],
dev_server_download_version="v1.7.4-standalone-nexus-operations",
)
elif env_type == "time-skipping":
Expand Down
1 change: 1 addition & 0 deletions tests/nexus_standalone_activity/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Tests for the Nexus standalone Activity sample."""
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import uuid
from datetime import timedelta

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

from nexus_standalone_activity.activity import create_greeting
from nexus_standalone_activity.handler import GreetingServiceHandler
from nexus_standalone_activity.service import (
GreetingInput,
GreetingOutput,
GreetingService,
)
from nexus_standalone_activity.worker import TASK_QUEUE
from tests.helpers.nexus import create_nexus_endpoint, delete_nexus_endpoint


async def test_nexus_operation_backed_by_standalone_activity(
client: Client, env: WorkflowEnvironment
) -> None:
if env.supports_time_skipping:
pytest.skip("Time-skipping server does not support standalone Nexus operations")

endpoint_name = f"test-nexus-standalone-activity-{uuid.uuid4()}"
create_response = await create_nexus_endpoint(
name=endpoint_name,
task_queue=TASK_QUEUE,
client=client,
)
try:
async with Worker(
client,
task_queue=TASK_QUEUE,
activities=[create_greeting],
nexus_service_handlers=[GreetingServiceHandler()],
):
nexus_client = client.create_nexus_client(
service=GreetingService,
endpoint=endpoint_name,
)
result = await nexus_client.execute_operation(
GreetingService.greet,
GreetingInput(name="Test"),
id=str(uuid.uuid4()),
schedule_to_close_timeout=timedelta(seconds=10),
)

assert isinstance(result, GreetingOutput)
assert result.message == "Hello, Test!"
finally:
_ = await delete_nexus_endpoint(
id=create_response.endpoint.id,
version=create_response.endpoint.version,
client=client,
)
Loading
Loading