1 - Getting started with the Dapr client Python SDK
How to get up and running with the Dapr Python SDK
The Dapr client package allows you to interact with other Dapr applications from a Python application.
Prerequisites
Install the Dapr Python package before getting started.
Import the client package
The dapr package contains the DaprClient, which is used to create and use a client.
from dapr.clients import DaprClient
Initialising the client
You can initialise a Dapr client in multiple ways:
Default values:
When you initialise the client without any parameters it will use the default values for a Dapr
sidecar instance (127.0.0.1:50001).
from dapr.clients import DaprClient
with DaprClient() as d:
# use the client
Specifying an endpoint on initialisation:
When passed as an argument in the constructor, the gRPC endpoint takes precedence over any
configuration or environment variable.
from dapr.clients import DaprClient
with DaprClient("mydomain:50051?tls=true") as d:
# use the client
Use the host:port?tls=true or host:port?tls=false form to set the endpoint and whether the connection uses TLS.
Deprecated
A gRPC endpoint that starts with http:// or https://, for example https://mydomain:443, still works but emits a DeprecationWarning that names the equivalent endpoint. Replace it with mydomain:443?tls=true (for https://) or mydomain:443?tls=false (for http://). Python hides a DeprecationWarning in many cases, so run with -W default to see it.Configuration options:
Dapr Sidecar Endpoints
You can use the standardised DAPR_GRPC_ENDPOINT environment variable to
specify the gRPC endpoint. When this variable is set, the client can be initialised
without any arguments:
export DAPR_GRPC_ENDPOINT="mydomain:50051?tls=true"
from dapr.clients import DaprClient
with DaprClient() as d:
# the client will use the endpoint specified in the environment variables
To connect to the sidecar over a Unix domain socket, give the absolute socket path after three slashes, for example unix:///tmp/dapr.sock. DaprClient, WorkflowRuntime and DaprWorkflowClient all accept this form. An endpoint with a host part after unix://, such as unix://dapr.sock, raises ValueError when the client is created. For a relative path, use unix:dapr.sock.
The legacy environment variables DAPR_RUNTIME_HOST, DAPR_HTTP_PORT and DAPR_GRPC_PORT are
also supported, but DAPR_GRPC_ENDPOINT takes precedence.
The http:// and https:// prefixes are deprecated for DAPR_GRPC_ENDPOINT too. Use host:port?tls=true or host:port?tls=false instead.
Dapr API Token
If your Dapr instance is configured to require the DAPR_API_TOKEN environment variable, you can
set it in the environment and the client will use it automatically.
You can read more about Dapr API token authentication here.
Health timeout
On client initialisation, a health check is performed against the Dapr sidecar (/healthz/outbound).
The client will wait for the sidecar to be up and running before proceeding.
The default healthcheck timeout is 60 seconds, but it can be overridden by setting the DAPR_HEALTH_TIMEOUT
environment variable. Each healthcheck attempt is also limited by this timeout, so a sidecar that accepts the
connection but does not answer cannot block the client forever.
Retries and timeout
The Dapr client can retry a request if a specific error code is received from the sidecar. This is
configurable through the DAPR_API_MAX_RETRIES environment variable and is picked up automatically,
not requiring any code changes.
The default value for DAPR_API_MAX_RETRIES is 0, which means no retries will be made.
You can fine-tune more retry parameters by creating a dapr.clients.retry.RetryPolicy object and
passing it to the DaprClient constructor:
from dapr.clients.retry import RetryPolicy
retry = RetryPolicy(
max_attempts=5,
initial_backoff=1,
max_backoff=20,
backoff_multiplier=1.5,
retryable_http_status_codes=[408, 429, 500, 502, 503, 504],
retryable_grpc_status_codes=[StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED, ]
)
with DaprClient(retry_policy=retry) as d:
...
or for actors:
factory = ActorProxyFactory(retry_policy=RetryPolicy(max_attempts=3))
proxy = ActorProxy.create('DemoActor', ActorId('1'), DemoActorInterface, factory)
Timeout can be set for all calls through the environment variable DAPR_API_TIMEOUT_SECONDS. By default, no timeout is set, so calls wait without a time limit.
Note: DAPR_API_TIMEOUT_SECONDS does not apply to the workflow client methods wait_for_workflow_start and wait_for_workflow_completion. Without their own timeout_in_seconds argument, they wait with no time limit. Pass timeout_in_seconds to limit the wait. See Dapr Workflow extension.
Note: You can control timeouts on service invocation separately, by passing a timeout parameter to the invoke_method method.
DNS resolution on macOS
On macOS, the client sets the GRPC_DNS_RESOLVER environment variable to native when it is not already set, so gRPC uses the operating system resolver. The resolver that ships with gRPC can fail with DNS query cancelled for host names that resolve on the system. To use a different resolver, set GRPC_DNS_RESOLVER before you create the client. The variable applies to the whole process.
Error handling
Initially, errors in Dapr followed the Standard gRPC error model. However, to provide more detailed and informative error messages, in version 1.13 an enhanced error model has been introduced which aligns with the gRPC Richer error model. In response, the Python SDK implemented DaprGrpcError, a custom exception class designed to improve the developer experience.
It’s important to note that the transition to using DaprGrpcError for all gRPC status exceptions is a work in progress. As of now, not every API call in the SDK has been updated to leverage this custom exception. We are actively working on this enhancement and welcome contributions from the community.
Example of handling DaprGrpcError exceptions when using the Dapr python-SDK:
try:
d.save_state(store_name=storeName, key=key, value=value)
except DaprGrpcError as err:
print(f'Status code: {err.code()}')
print(f"Message: {err.message()}")
print(f"Error code: {err.error_code()}")
print(f"Error info(reason): {err.error_info.reason}")
print(f"Resource info (resource type): {err.resource_info.resource_type}")
print(f"Resource info (resource name): {err.resource_info.resource_name}")
print(f"Bad request (field): {err.bad_request.field_violations[0].field}")
print(f"Bad request (description): {err.bad_request.field_violations[0].description}")
Building blocks
The Python SDK allows you to interface with all of the Dapr building blocks.
Invoke a service
The Dapr Python SDK provides a simple API for invoking services via either HTTP or gRPC (deprecated). The protocol can be selected by setting the DAPR_API_METHOD_INVOCATION_PROTOCOL environment variable, defaulting to HTTP when unset. GRPC service invocation in Dapr is deprecated and GRPC proxying is recommended as an alternative.
from dapr.clients import DaprClient
with DaprClient() as d:
# invoke a method (gRPC or HTTP GET)
resp = d.invoke_method('service-to-invoke', 'method-to-invoke', data='{"message":"Hello World"}')
# for other HTTP verbs the verb must be specified
# invoke a 'POST' method (HTTP only)
resp = d.invoke_method('service-to-invoke', 'method-to-invoke', data='{"id":"100", "FirstName":"Value", "LastName":"Value"}', http_verb='post')
The base endpoint for HTTP api calls is specified in the DAPR_HTTP_ENDPOINT environment variable.
If this variable is not set, the endpoint value is derived from the DAPR_RUNTIME_HOST and DAPR_HTTP_PORT variables, whose default values are 127.0.0.1 and 3500 accordingly.
The base endpoint for gRPC calls is the one used for the client initialisation (explained above).
Save & get application state
from dapr.clients import DaprClient
with DaprClient() as d:
# Save state
d.save_state(store_name="statestore", key="key1", value="value1")
# Get state
data = d.get_state(store_name="statestore", key="key1").data
# Delete state
d.delete_state(store_name="statestore", key="key1")
Query application state (Alpha)
from dapr import DaprClient
query = '''
{
"filter": {
"EQ": { "state": "CA" }
},
"sort": [
{
"key": "person.id",
"order": "DESC"
}
]
}
'''
with DaprClient() as d:
resp = d.query_state(
store_name='state_store',
query=query,
states_metadata={"metakey": "metavalue"}, # optional
)
Publish & subscribe
Publish messages
from dapr.clients import DaprClient
with DaprClient() as d:
resp = d.publish_event(pubsub_name='pubsub', topic_name='TOPIC_A', data='{"message":"Hello World"}')
Send CloudEvents messages with a json payload:
from dapr.clients import DaprClient
import json
with DaprClient() as d:
cloud_event = {
'specversion': '1.0',
'type': 'com.example.event',
'source': 'my-service',
'id': 'myid',
'data': {'id': 1, 'message': 'hello world'},
'datacontenttype': 'application/json',
}
# Set the data content type to 'application/cloudevents+json'
resp = d.publish_event(
pubsub_name='pubsub',
topic_name='TOPIC_CE',
data=json.dumps(cloud_event),
data_content_type='application/cloudevents+json',
)
Publish CloudEvents messages with plain text payload:
from dapr.clients import DaprClient
import json
with DaprClient() as d:
cloud_event = {
'specversion': '1.0',
'type': 'com.example.event',
'source': 'my-service',
'id': "myid",
'data': 'hello world',
'datacontenttype': 'text/plain',
}
# Set the data content type to 'application/cloudevents+json'
resp = d.publish_event(
pubsub_name='pubsub',
topic_name='TOPIC_CE',
data=json.dumps(cloud_event),
data_content_type='application/cloudevents+json',
)
Publish several messages in one call with publish_events. Each item in data can be a str, bytes, or a BulkPublishEntry. Use BulkPublishEntry to set metadata, a content type, or an entry ID on one message. The publish_metadata applies to every message, and individual entry metadata overrides it:
from dapr.clients import BulkPublishEntry, DaprClient
import json
with DaprClient() as d:
resp = d.publish_events(
pubsub_name='pubsub',
topic_name='TOPIC_A',
data=[
json.dumps({'id': 1, 'message': 'hello world'}),
BulkPublishEntry(
event=json.dumps({'id': 2, 'message': 'hello tenant-a'}),
metadata={'partitionKey': 'tenant-a'},
),
],
data_content_type='application/json',
publish_metadata={'ttlInSeconds': '60'},
)
for entry in resp.failed_entries:
print(f'failed entry {entry.entry_id}: {entry.error}')
For more details, see the bulk publish guide.
Subscribe to messages
from dapr.ext.grpc import App, Rule, SubscriptionMessage
app = App()
# Default subscription for a topic
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A')
def mytopic(event: SubscriptionMessage) -> None:
data = event.data()
print(f'Received: id={data["id"]}, message="{data ["message"]}"'
f' content_type="{event.data_content_type()}"',flush=True)
# Specific handler using Pub/Sub routing
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A',
rule=Rule("event.type == \"important\"", 1))
def mytopic_important(event: SubscriptionMessage) -> None:
data = event.data()
print(f'Received: id={data["id"]}, message="{data ["message"]}"'
f' content_type="{event.data_content_type()}"',flush=True)
Annotate the event parameter with SubscriptionMessage, the same type that streaming subscriptions deliver. data() returns the payload parsed by content type, for example a dict for JSON. Handlers that are not annotated, or that are annotated with cloudevents.sdk.event.v1.Event, receive the deprecated CloudEvents-based event, and subscribe() emits a DeprecationWarning. See Python gRPC extension for the deprecation timeline.
Streaming message subscription
You can create a streaming subscription to a PubSub topic using either the subscribe
or subscribe_handler methods.
The subscribe method returns an iterable Subscription object, which allows you to pull messages from the
stream by using a for loop (ex. for message in subscription) or by
calling the next_message method. This will block on the main thread while waiting for messages.
When done, you should call the close method to terminate the
subscription and stop receiving messages.
If the stream fails with a transient error or the sidecar ends it, the subscription reconnects and returns the next message. Before Python SDK 1.19, the subscription returned None after a reconnect, so you no longer need to check for None.
After five failed reconnect attempts, next_message() and the for loop raise StreamReconnectError. Catch this error to decide whether to create a new subscription or to stop the application.
The subscribe_with_handler method accepts a callback function that is executed for each message
received from the stream.
It runs in a separate thread, so it doesn’t block the main thread. The callback should return a
TopicEventResponse (ex. TopicEventResponse('success')), indicating whether the message was
processed successfully, should be retried, or should be discarded. The method will automatically
manage message acknowledgements based on the returned status. The call to subscribe_with_handler
method returns a close function, which should be called to terminate the subscription when you’re
done.
Here’s an example of using the subscribe method:
import time
from dapr.clients import DaprClient
from dapr.clients.grpc.subscription import (
StreamCancelledError,
StreamInactiveError,
StreamReconnectError,
)
counter = 0
def process_message(message):
global counter
counter += 1
# Process the message here
print(f'Processing message: {message.data()} from {message.topic()}...')
return 'success'
def main():
with DaprClient() as client:
global counter
subscription = client.subscribe(
pubsub_name='pubsub', topic='TOPIC_A', dead_letter_topic='TOPIC_A_DEAD'
)
try:
for message in subscription:
try:
response_status = process_message(message)
if response_status == 'success':
subscription.respond_success(message)
elif response_status == 'retry':
subscription.respond_retry(message)
elif response_status == 'drop':
subscription.respond_drop(message)
if counter >= 5:
break
except StreamInactiveError:
print('Stream is inactive. Retrying...')
time.sleep(1)
continue
except StreamCancelledError:
print('Stream was cancelled')
break
except Exception as e:
print(f'Error occurred during message processing: {e}')
except StreamReconnectError:
print('Stream could not be reconnected after 5 attempts')
finally:
print('Closing subscription...')
subscription.close()
if __name__ == '__main__':
main()
And here’s an example of using the subscribe_with_handler method:
import time
from dapr.clients import DaprClient
from dapr.clients.grpc._response import TopicEventResponse
counter = 0
def process_message(message):
# Process the message here
global counter
counter += 1
print(f'Processing message: {message.data()} from {message.topic()}...')
return TopicEventResponse('success')
def main():
with (DaprClient() as client):
# This will start a new thread that will listen for messages
# and process them in the `process_message` function
close_fn = client.subscribe_with_handler(
pubsub_name='pubsub', topic='TOPIC_A', handler_fn=process_message,
dead_letter_topic='TOPIC_A_DEAD'
)
while counter < 5:
time.sleep(1)
print("Closing subscription...")
close_fn()
if __name__ == '__main__':
main()
Conversation (Alpha)
Note
The Dapr Conversation API is currently in alpha.Since version 1.15 Dapr offers developers the capability to securely and reliably interact with Large Language Models (LLM) through the Conversation API.
from dapr.clients import DaprClient
from dapr.clients.grpc.conversation import ConversationInput
with DaprClient() as d:
inputs = [
ConversationInput(content="What's Dapr?", role='user', scrub_pii=True),
ConversationInput(content='Give a brief overview.', role='user', scrub_pii=True),
]
metadata = {
'model': 'foo',
'key': 'authKey',
'cacheTTL': '10m',
}
response = d.converse_alpha1(
name='echo', inputs=inputs, temperature=0.7, context_id='chat-123', metadata=metadata
)
for output in response.outputs:
print(f'Result: {output.result}')
Interact with output bindings
from dapr.clients import DaprClient
with DaprClient() as d:
resp = d.invoke_binding(binding_name='kafkaBinding', operation='create', data='{"message":"Hello World"}')
Retrieve secrets
from dapr.clients import DaprClient
with DaprClient() as d:
resp = d.get_secret(store_name='localsecretstore', key='secretKey')
Configuration
Get configuration
from dapr.clients import DaprClient
with DaprClient() as d:
# Get Configuration
configuration = d.get_configuration(store_name='configurationstore', keys=['orderId'], config_metadata={})
Subscribe to configuration
import asyncio
from time import sleep
from dapr.clients import DaprClient
async def executeConfiguration():
with DaprClient() as d:
storeName = 'configurationstore'
key = 'orderId'
# Wait for sidecar to be up within 20 seconds.
d.wait(20)
# Subscribe to configuration by key.
configuration = await d.subscribe_configuration(store_name=storeName, keys=[key], config_metadata={})
while True:
if configuration != None:
items = configuration.get_items()
for key, item in items:
print(f"Subscribe key={key} value={item.value} version={item.version}", flush=True)
else:
print("Nothing yet")
sleep(5)
asyncio.run(executeConfiguration())
Distributed Lock
from dapr.clients import DaprClient
def main():
# Lock parameters
store_name = 'lockstore' # as defined in components/lockstore.yaml
resource_id = 'example-lock-resource'
client_id = 'example-client-id'
expiry_in_seconds = 60
with DaprClient() as dapr:
print('Will try to acquire a lock from lock store named [%s]' % store_name)
print('The lock is for a resource named [%s]' % resource_id)
print('The client identifier is [%s]' % client_id)
print('The lock will will expire in %s seconds.' % expiry_in_seconds)
with dapr.try_lock(store_name, resource_id, client_id, expiry_in_seconds) as lock_result:
assert lock_result.success, 'Failed to acquire the lock. Aborting.'
print('Lock acquired successfully!!!')
# At this point the lock was released - by magic of the `with` clause ;)
unlock_result = dapr.unlock(store_name, resource_id, client_id)
print('We already released the lock so unlocking will not work.')
print('We tried to unlock it anyway and got back [%s]' % unlock_result.status)
Cryptography
from dapr.clients import DaprClient
message = 'The secret is "passw0rd"'
def main():
with DaprClient() as d:
resp = d.encrypt(
data=message.encode(),
options=EncryptOptions(
component_name='crypto-localstorage',
key_name='rsa-private-key.pem',
key_wrap_algorithm='RSA',
),
)
encrypt_bytes = resp.read()
resp = d.decrypt(
data=encrypt_bytes,
options=DecryptOptions(
component_name='crypto-localstorage',
key_name='rsa-private-key.pem',
),
)
decrypt_bytes = resp.read()
print(decrypt_bytes.decode()) # The secret is "passw0rd"
Python SDK examples
2 - Getting started with the Dapr actor Python SDK
How to get up and running with the Dapr Python SDK
The Dapr actor package allows you to interact with Dapr virtual actors from a Python application.
Pre-requisites
Actor interface
The interface defines the actor contract that is shared between the actor implementation and the clients calling the actor. Because a client may depend on it, it typically makes sense to define it in an assembly that is separate from the actor implementation.
from dapr.actor import ActorInterface, actormethod
class DemoActorInterface(ActorInterface):
@actormethod(name="GetMyData")
async def get_my_data(self) -> object:
...
Actor services
An actor service hosts the virtual actor. It is implemented a class that derives from the base type Actor and implements the interfaces defined in the actor interface.
Actors can be created using one of the Dapr actor extensions:
Actor client
An actor client contains the implementation of the actor client which calls the actor methods defined in the actor interface.
import asyncio
from dapr.actor import ActorProxy, ActorId
from demo_actor_interface import DemoActorInterface
async def main():
# Create proxy client
proxy = ActorProxy.create('DemoActor', ActorId('1'), DemoActorInterface)
# Call method on client
resp = await proxy.GetMyData()
Sample
Visit this page for a runnable actor sample.
Mock Actor Testing
The Dapr Python SDK provides the ability to create mock actors to unit test your actor methods and see how they interact with the actor state.
Sample Usage
from dapr.actor.runtime.mock_actor import create_mock_actor
class MyActor(Actor, MyActorInterface):
async def save_state(self, data) -> None:
await self._state_manager.set_state('mystate', data)
await self._state_manager.save_state()
mock_actor = create_mock_actor(MyActor, "id")
await mock_actor.save_state(5)
assert mockactor._state_manager._mock_state['mystate'] == 5 #True
Mock actors are created by passing your actor class and an actor ID (a string) to the create_mock_actor function. This function returns an instance of the actor with many internal methods overridden. Instead of interacting with Dapr for tasks like saving state or managing timers, the mock actor uses in-memory state to simulate these behaviors.
This state can be accessed through the following variables:
IMPORTANT NOTE: Due to type hinting issues as discussed further down, these variables will not be visible to type hinters/linters/etc, who will think they are invalid variables. You will need to use them with #type: ignore in order to satisfy any such systems.
_state_manager._mock_state()
A [str, object] dict where all the actor state is stored. Any variable saved via _state_manager.save_state(key, value), or any other statemanager method is stored in the dict as that key, value pair. Any value loaded via try_get_state or any other statemanager method is taken from this dict.
_state_manager._mock_timers()
A [str, ActorTimerData] dict which holds the active actor timers. Any actor method which would add or remove a timer adds or pops the appropriate ActorTimerData object from this dict.
_state_manager._mock_reminders()
A [str, ActorReminderData] dict which holds the active actor reminders. Any actor method which would add or remove a timer adds or pops the appropriate ActorReminderData object from this dict.
Note: The timers and reminders will never actually trigger. The dictionaries exist only so methods that should add or remove timers/reminders can be tested. If you need to test the callbacks they should activate, you should call them directly with the appropriate values:
result = await mock_actor.recieve_reminder(name, state, due_time, period, _ttl)
# Test the result directly or test for side effects (like changing state) by querying `_state_manager._mock_state`
Usage and Limitations
To allow for more fine-grained control, the _on_activate method will not be called automatically the way it is when Dapr initializes a new Actor instance. You should call it manually as needed as part of your tests.
A current limitation of the mock actor system is that it does not call the _on_pre_actor_method and _on_post_actor_method methods. You can always call these methods manually as part of a test.
The __init__, register_timer, unregister_timer, register_reminder, unregister_reminder methods are all overwritten by the MockActor class that gets applied as a mixin via create_mock_actor. If your actor itself overwrites these methods, those modifications will themselves be overwritten and the actor will likely not behave as you expect.
note: __init__ is a special case where you are expected to define it as
def __init__(self, ctx, actor_id):
super().__init__(ctx, actor_id)
Mock actors work fine with this, but if you have added any extra logic into __init__, it will be overwritten. It is worth noting that the correct way to apply logic on initialization is via _on_activate (which can also be safely used with mock actors) instead of __init__.
If you have an actor which does override default Dapr actor methods, you can create a custom subclass of the MockActor class (from MockActor.py) which implements whatever custom logic you have along with interacting with _mock_state, _mock_timers, and _mock_reminders as normal, and then applying that custom class as a mixin via a create_mock_actor function you define yourself.
The actor _runtime_ctx variable is set to None. All the normal actor methods have been overwritten such as to not call it, but if your code itself interacts directly with _runtime_ctx, tests may fail.
The actor _state_manager is overwritten with an instance of MockStateManager. This has all the same methods and functionality of the base ActorStateManager, except for using the various _mock variables for storing data instead of the _runtime_ctx. If your code implements its own custom state manager it will be overwritten and tests will likely fail.
Type Hinting
Because of Python’s lack of a unified method for type hinting type intersections (see: python/typing #213), type hinting unfortunately doesn’t work with Mock Actors. The return type is type hinted as “instance of Actor subclass T” when it should really be type hinted as “instance of MockActor subclass T” or “instance of type intersection [Actor subclass T, MockActor]” (where, it is worth noting, MockActor is itself a subclass of Actor).
This means that, for example, if you hover over mockactor._state_manager in a code editor, it will come up as an instance of ActorStateManager (instead of MockStateManager), and various IDE helper functions (like VSCode’s Go to Definition, which will bring you to the definition of ActorStateManager instead of MockStateManager) won’t work properly.
For now, this issue is unfixable, so it’s merely something to be noted because of the confusion it might cause. If in the future it becomes possible to accurately type hint cases like this feel free to open an issue about implementing it.
3.1 - Getting started with the Dapr Python gRPC service extension
How to get up and running with the Dapr Python gRPC extension
The Dapr Python SDK provides a built in gRPC server extension, dapr.ext.grpc, for creating Dapr services.
Installation
You can download and install the Dapr gRPC server extension with:
Note
The development version will contain features and behavior that will be compatible with the pre-release version of the Dapr runtime. Only tagged releases are published to PyPI, so install the development version from the GitHub repository. Make sure to uninstall any stable versions of the Python SDK before installing it.
pip install "dapr[grpc] @ git+https://github.com/dapr/python-sdk.git@main"
Starting with Dapr 1.19, this extension is an extra of the dapr package. The dapr-ext-grpc package is no longer published. If you installed it before, follow the upgrade steps to move to the extra. The import path does not change.
Examples
The App object can be used to create a server.
Listen for service invocation requests
The InvokeMethodReqest and InvokeMethodResponse objects can be used to handle incoming requests.
A simple service that will listen and respond to requests will look like:
from dapr.ext.grpc import App, InvokeMethodRequest, InvokeMethodResponse
app = App()
@app.method(name='my-method')
def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse:
print(request.metadata, flush=True)
print(request.text(), flush=True)
return InvokeMethodResponse(b'INVOKE_RECEIVED', "text/plain; charset=UTF-8")
app.run(50051)
A full sample can be found here.
Subscribe to a topic
When subscribing to a topic, you can instruct dapr whether the event delivered has been accepted, or whether it should be dropped, or retried later.
Annotate the event parameter of each handler with SubscriptionMessage, the same type that streaming subscriptions deliver. It gives you the parsed payload (data()), the delivery metadata (metadata()) and the CloudEvent fields, such as id(), type() and extensions().
from typing import Optional
from dapr.ext.grpc import App, Rule, SubscriptionMessage
from dapr.clients.grpc._response import TopicEventResponse
app = App()
# Default subscription for a topic
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A')
def mytopic(event: SubscriptionMessage) -> Optional[TopicEventResponse]:
print(event.data(),flush=True)
# Returning None (or not doing a return explicitly) is equivalent
# to returning a TopicEventResponse("success").
# You can also return TopicEventResponse("retry") for dapr to log
# the message and retry delivery later, or TopicEventResponse("drop")
# for it to drop the message
return TopicEventResponse("success")
# Specific handler using Pub/Sub routing
@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A',
rule=Rule("event.type == \"important\"", 1))
def mytopic_important(event: SubscriptionMessage) -> None:
print(event.data(),flush=True)
# Handler with disabled topic validation
@app.subscribe(pubsub_name='pubsub-mqtt', topic='topic/#', disable_topic_validation=True,)
def mytopic_wildcard(event: SubscriptionMessage) -> None:
print(event.data(),flush=True)
app.run(50051)
A full sample can be found here.
Deprecated
Handlers that are not annotated, or that are annotated with cloudevents.sdk.event.v1.Event, receive the CloudEvents-based event, and subscribe() emits a DeprecationWarning when it registers them. Annotate them with SubscriptionMessage before the type is removed:
- In 1.20, handlers that are not annotated receive
SubscriptionMessage. Only handlers annotated with the CloudEvents type still receive it. - In 1.21, the
dapr[grpc] extra no longer installs cloudevents. - In 1.22, the CloudEvents-based type is removed.
from dapr.ext.grpc import App, BindingRequest
app = App()
@app.binding('kafkaBinding')
def binding(request: BindingRequest):
print(request.text(), flush=True)
app.run(50051)
A full sample can be found here.
3.4 - Dapr Python SDK integration with Dapr Workflow extension
How to get up and running with the Dapr Workflow extension
The Dapr Python SDK provides a built-in Dapr Workflow extension, dapr.ext.workflow, for creating Dapr services.
Installation
You can download and install the Dapr Workflow extension with:
pip install "dapr[workflow]"
Note
The development version will contain features and behavior that will be compatible with the pre-release version of the Dapr runtime. Only tagged releases are published to PyPI, so install the development version from the GitHub repository. Make sure to uninstall any stable versions of the Python SDK before installing it.
pip install "dapr[workflow] @ git+https://github.com/dapr/python-sdk.git@main"
Starting with Dapr 1.19, this extension is an extra of the dapr package. The dapr-ext-workflow package is no longer published. If you installed it before, follow the upgrade steps to move to the extra. The import path does not change.
Example
from time import sleep
import dapr.ext.workflow as wf
wfr = wf.WorkflowRuntime()
@wfr.workflow(name='random_workflow')
def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
try:
result1 = yield ctx.call_activity(step1, input=wf_input)
result2 = yield ctx.call_activity(step2, input=result1)
except Exception as e:
yield ctx.call_activity(error_handler, input=str(e))
raise
return [result1, result2]
@wfr.activity(name='step1')
def step1(ctx, activity_input):
print(f'Step 1: Received input: {activity_input}.')
# Do some work
return activity_input + 1
@wfr.activity
def step2(ctx, activity_input):
print(f'Step 2: Received input: {activity_input}.')
# Do some work
return activity_input * 2
@wfr.activity
def error_handler(ctx, error):
print(f'Executing error handler: {error}.')
# Do some compensating work
if __name__ == '__main__':
wfr.start()
sleep(10) # wait for workflow runtime to start
wf_client = wf.DaprWorkflowClient()
instance_id = wf_client.schedule_new_workflow(workflow=task_chain_workflow, input=42)
print(f'Workflow started. Instance ID: {instance_id}')
state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
print(f'Workflow completed! Status: {state.runtime_status}')
wfr.shutdown()
- Learn more about authoring and managing workflows:
- Visit Python SDK examples for code samples and instructions to try out Dapr Workflow:
Stateful workflow history
From Dapr 1.19, the workflow worker keeps the history of each workflow instance in memory, so the Dapr sidecar sends only the new events on each turn. This is on by default. To turn it off, for example when worker memory is limited, set disable_stateful_history=True on WorkflowRuntime:
import dapr.ext.workflow as wf
wfr = wf.WorkflowRuntime(disable_stateful_history=True)
Learn more in Stateful workflow client.
Waiting for workflows
wait_for_workflow_start and wait_for_workflow_completion ignore the DAPR_API_TIMEOUT_SECONDS environment variable. Without the timeout_in_seconds argument, they wait with no time limit. Pass timeout_in_seconds to limit how long the call waits. When the time limit passes, the call raises TimeoutError:
state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
Next steps
Getting started with the Dapr Workflow Python SDK3.4.1 - Getting started with the Dapr Workflow Python SDK
How to get up and running with workflows using the Dapr Python SDK
Let’s create a Dapr workflow and invoke it using the console. With the provided workflow example, you will:
- Run a Python console application that demonstrates workflow orchestration with activities, child workflows, and external events
- Learn how to handle retries, timeouts, and workflow state management
- Use the Python workflow SDK to start, pause, resume, and purge workflow instances
This example uses the default configuration from dapr init in self-hosted mode.
In the Python example project, the simple.py file contains the setup of the app, including:
- The workflow definition
- The workflow activity definitions
- The registration of the workflow and workflow activities
Prerequisites
Set up the environment
Start by cloning the [Python SDK repo].
git clone https://github.com/dapr/python-sdk.git
From the Python SDK root directory, navigate to the Dapr Workflow example.
Run the following command to install the requirements for running this workflow sample with the Dapr Python SDK.
pip3 install -r workflow/requirements.txt
Run the application locally
To run the Dapr application, you need to start the Python program and a Dapr sidecar. In the terminal, run:
dapr run --app-id wf-simple-example --dapr-grpc-port 50001 --resources-path components -- python3 simple.py
Note: Since Python3.exe is not defined in Windows, you may need to use python simple.py instead of python3 simple.py.
Expected output
- "== APP == Hi Counter!"
- "== APP == New counter value is: 1!"
- "== APP == New counter value is: 11!"
- "== APP == Retry count value is: 0!"
- "== APP == Retry count value is: 1! This print statement verifies retry"
- "== APP == Appending 1 to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending 2 to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending 3 to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Get response from hello_world_wf after pause call: Suspended"
- "== APP == Get response from hello_world_wf after resume call: Running"
- "== APP == New counter value is: 111!"
- "== APP == New counter value is: 1111!"
- "== APP == Workflow completed! Result: "Completed"
What happened?
When you run the application, several key workflow features are shown:
Workflow and Activity Registration: The application uses Python decorators to automatically register workflows and activities with the runtime. This decorator-based approach provides a clean, declarative way to define your workflow components:
@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
# Workflow definition...
@wfr.activity(name='hello_act')
def hello_act(ctx: WorkflowActivityContext, wf_input):
# Activity definition...
Runtime Setup: The application initializes the workflow runtime and client:
wfr = WorkflowRuntime()
wfr.start()
wf_client = DaprWorkflowClient()
Activity Execution: The workflow executes a series of activities that increment a counter:
@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
yield ctx.call_activity(hello_act, input=1)
yield ctx.call_activity(hello_act, input=10)
Retry Logic: The workflow demonstrates error handling with a retry policy:
retry_policy = RetryPolicy(
first_retry_interval=timedelta(seconds=1),
max_number_of_attempts=3,
backoff_coefficient=2,
max_retry_interval=timedelta(seconds=10),
retry_timeout=timedelta(seconds=100),
)
yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
Child Workflow: A child workflow is executed with its own retry policy:
yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)
External Event Handling: The workflow waits for an external event with a timeout:
event = ctx.wait_for_external_event(event_name)
timeout = ctx.create_timer(timedelta(seconds=30))
winner = yield when_any([event, timeout])
Workflow Lifecycle Management: The example demonstrates how to pause and resume the workflow:
wf_client.pause_workflow(instance_id=instance_id)
metadata = wf_client.get_workflow_state(instance_id=instance_id)
# ... check status ...
wf_client.resume_workflow(instance_id=instance_id)
Event Raising: After resuming, the workflow raises an event:
wf_client.raise_workflow_event(
instance_id=instance_id,
event_name=event_name,
data=event_data
)
Completion and Cleanup: Finally, the workflow waits for completion and cleans up:
state = wf_client.wait_for_workflow_completion(
instance_id,
timeout_in_seconds=30
)
wf_client.purge_workflow(instance_id=instance_id)
List, inspect and rerun workflow instances
From Dapr 1.19, DaprWorkflowClient and the async client in dapr.ext.workflow.aio can list the workflow instances of the app, read the history of an instance, and rerun a finished instance from one of its history events. Use them to find instances without storing their IDs yourself, to see where a failed instance stopped, and to retry it from that step with a corrected input instead of from the start.
import dapr.ext.workflow as wf
wf_client = wf.DaprWorkflowClient()
# Page through the instance IDs of this app, 100 IDs per request
for instance_id in wf_client.iter_workflow_instance_ids(page_size=100):
print(instance_id)
# Read the history of an instance, oldest event first
history = wf_client.get_workflow_history('order-123')
for event in history:
print(event.event_id, event.event_type.name, event.name)
# Rerun from the scheduling of the charge_order activity, with a new input
event_id = next(e.event_id for e in history if e.is_rerunnable and e.name == 'charge_order')
new_instance_id = wf_client.rerun_workflow_from_event('order-123', event_id, input=25)
list_workflow_instance_ids(page_size=..., continuation_token=...) returns one page with instance_ids and continuation_token. iter_workflow_instance_ids() fetches the pages for you. Listing needs an actor state store that can list keys. Otherwise the call raises NotImplementedError.rerun_workflow_from_event starts a new instance and leaves the source instance unchanged. The source instance must be in a terminal state and must not be a child workflow. The event must be the scheduling of an activity, the creation of a timer or the creation of a child workflow, which is_rerunnable reports. Omit input to keep the recorded input, and pass new_instance_id to choose the ID of the new instance.
For a runnable version, see the workflow management example.
Large workflow payloads
By default, the workflow gRPC channel receives messages of up to 4 MiB, the gRPC default, so a larger message, for example a large workflow input or output, fails with RESOURCE_EXHAUSTED. To raise the limit for sending and receiving, pass max_grpc_message_length in bytes to WorkflowRuntime, DaprWorkflowClient and the async DaprWorkflowClient, or set the DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES environment variable. The argument takes precedence over the environment variable.
wfr = wf.WorkflowRuntime(max_grpc_message_length=16 * 1024 * 1024)
wf_client = wf.DaprWorkflowClient(max_grpc_message_length=16 * 1024 * 1024)
Worker readiness
wfr.start() waits until the worker has opened its work-item stream to the Dapr sidecar, for 30 seconds by default (worker_ready_timeout on WorkflowRuntime). After that, wfr.wait_for_worker_ready(timeout=30.0) returns True when the stream is open, and False if the stream does not open again within the timeout, for example while the worker reconnects after a sidecar restart. Use it in a readiness check, so that the app reports not ready while its worker cannot receive work. Before Python SDK 1.19, it kept returning True after the first connection.
Next steps
3.5.1 - Getting started with the Dapr Databricks Python SDK extension
How to turn Databricks Lakeflow streaming records into durable Dapr Workflow executions
Let’s run a fraud-remediation example that turns simulated Databricks Lakeflow records into durable
Dapr Workflow executions. With the
provided example, you will:
- Run a Dapr Workflow (
fraud_remediation) that freezes a card, notifies the customer, opens an
investigation, waits for an analyst decision, and resolves the case - Feed it two simulated Lakeflow micro-batches through
dapr.ext.databricks’s batch handler — no
Databricks workspace or pyspark required for this part - See a retried micro-batch schedule zero duplicate workflow executions, because every instance ID
is derived deterministically from the record’s own business key
Databricks Lakeflow identifies a suspicious transaction
|
dapr.ext.databricks sink
|
FraudRemediationWorkflow
|
freeze card -> notify customer -> open investigation
|
wait for analyst decision
|
resolve case
This example uses the default configuration from dapr init in
self-hosted mode.
Prerequisites
Set up the environment
Clone the Python SDK repo:
git clone https://github.com/dapr/python-sdk.git
From the repo root, go to the Databricks example:
This example has two files:
fraud_remediation_workflow.py — the Dapr Workflow side: the fraud_remediation workflow and
its activities. Its __main__ block simulates a small Lakeflow micro-batch in-process using
DaprWorkflowBatchHandler directly — the same building block register_workflow_sink uses
internally — then retries the identical batch to demonstrate that no duplicate workflow
executions are created. This is the file you’ll run below.fraud_remediation_pipeline.py — the Databricks Lakeflow side: register_workflow_sink plus
the @dp.append_flow that feeds it. This file is illustrative only — it depends on
pyspark.pipelines (provided by the Databricks Lakeflow runtime, not pip install pyspark) and a
Unity Catalog table, so it can’t run locally. Copy its pattern into an actual Databricks Lakeflow
pipeline to wire the integration up for real.
Run the application locally
To run the example, start the Python program and a Dapr sidecar together:
dapr run --app-id fraud-remediation-demo -- python3 fraud_remediation_workflow.py
Expected output (interleaved with Dapr/durabletask log lines):
*** Lakeflow micro-batch 1: scheduling fraud remediation workflows
*** Micro-batch 1 handed off; each workflow now runs independently
*** Simulating a Lakeflow retry of the same micro-batch
*** Retry complete: no duplicate workflow executions were created
*** Triggered by sink 'fraud_actions', batch 1
*** Freezing card for transaction T-1001 (customer C-1)
*** Notifying customer C-1 about the frozen card
*** Opened investigation CASE-T-1001
*** Resolved CASE-T-1001: CONFIRMED_FRAUD
*** Workflow fraud-fraud_actions-v1-T-1001 completed: {"case_id": "CASE-T-1001", "decision": "CONFIRMED_FRAUD"}
... (and the same for T-1002)
Look for the structured dapr.ext.databricks: sink=... outcome=... log lines: the first
micro-batch reports outcome=newly_scheduled for both transactions, and the simulated retry
reports outcome=already_existed for both — the same transaction never starts a second
remediation workflow.
When you’re done:
dapr stop --app-id fraud-remediation-demo
What happened?
register_workflow_sink (used inside the example’s batch handler, and the function you’d
call from a real @dp.append_flow pipeline) computes a deterministic workflow instance ID for
every record — <namespace>-<sink>-<generation>-<business_key> — never a random UUID.- For each record, the handler checks whether a workflow instance with that ID already exists. If
not, it schedules one; if so, it treats the record as already handled. Both the first micro-batch
and the simulated retry go through this same check, which is why the retry logs
outcome=already_existed instead of starting a second fraud_remediation execution. - Once scheduled, each workflow instance runs independently of the Lakeflow batch that created
it —
freeze card -> notify customer -> open investigation -> wait for an analyst decision -> resolve case continues even after the streaming micro-batch has moved on. - If neither the “does it exist” check nor the schedule call can complete (Dapr unavailable, a
timeout, an auth failure), the whole micro-batch fails so Lakeflow retries it — the sink never
reports success while a record’s fate is unknown.
This is retry-safe, not exactly-once: it depends on Dapr retaining that instance’s
history/state. See the
delivery semantics and full-refresh section
of the extension’s reference docs for what that means for your workflow-history retention, and how
generation lets you deliberately replay business actions after a full Lakeflow pipeline refresh.
Next steps
- Copy
fraud_remediation_pipeline.py’s pattern into a real Databricks Lakeflow pipeline to wire
this up against real streaming data. - See the
full configuration reference —
composite business keys, custom input mapping, bounded concurrency, and observability.
- Learn more about authoring the workflow side: Dapr Workflow.
4 -
The Dapr Conversation API is currently in alpha. This page presents the recommended, minimal patterns to use it effectively with the Python SDK:
- Plain requests (no tools)
- Requests with tools (functions as tools)
- Multi‑turn flows with tool execution
- Async variants
- Important safety notes for executing tool calls
Prerequisites
For full, end‑to‑end flows and provider setup, see:
- The SDK examples under Conversation:
from dapr.clients import DaprClient
from dapr.clients.grpc import conversation
# Build a single‑turn Alpha2 input
user_msg = conversation.create_user_message("What's Dapr?")
alpha2_input = conversation.ConversationInputAlpha2(messages=[user_msg])
with DaprClient() as client:
resp = client.converse_alpha2(
name="echo", # replace with your LLM component name
inputs=[alpha2_input],
temperature=1,
)
for msg in resp.to_assistant_messages():
if msg.of_assistant.content:
print(msg.of_assistant.content[0].text)
Key points:
- Use
conversation.create_user_message to build messages. - Wrap into
ConversationInputAlpha2(messages=[...]) and pass to converse_alpha2. - Use
response.to_assistant_messages() to iterate assistant outputs.
Decorator-based tools offer a clean, ergonomic approach. Define a function with clear type hints and detail docstring, this is important for the LLM to understand how or when to invoke the tool;
decorate it with @conversation.tool. Registered tools can be passed to the LLM and invoked via tool calls.
from dapr.clients import DaprClient
from dapr.clients.grpc import conversation
@conversation.tool
def get_weather(location: str, unit: str = 'fahrenheit') -> str:
"""Get current weather for a location."""
# Replace with a real implementation
return f"Weather in {location} (unit={unit})"
user_msg = conversation.create_user_message("What's the weather in Paris?")
alpha2_input = conversation.ConversationInputAlpha2(messages=[user_msg])
with DaprClient() as client:
response = client.converse_alpha2(
name="openai", # your LLM component
inputs=[alpha2_input],
tools=conversation.get_registered_tools(), # tools registered by @conversation.tool
tool_choice='auto',
temperature=1,
)
# Inspect assistant messages, including any tool calls
for msg in response.to_assistant_messages():
if msg.of_assistant.tool_calls:
for tc in msg.of_assistant.tool_calls:
print(f"Tool call: {tc.function.name} args={tc.function.arguments}")
elif msg.of_assistant.content:
print(msg.of_assistant.content[0].text)
Notes:
- Use
conversation.get_registered_tools() to collect all @conversation.tool decorated functions. - The binder validates/coerces params using your function signature. Keep annotations accurate.
This is the go‑to loop for tool‑using conversations:
Warning
Do not blindly auto‑execute tool calls returned by the LLM unless you trust all tools registered. Treat tool names and arguments as untrusted input.
- Validate inputs and enforce guardrails (allow‑listed tools, argument schemas, side‑effect constraints).
- For async or I/O‑bound tools, prefer
conversation.execute_registered_tool_async(..., timeout=...) and set conservative timeouts. - Consider adding a policy layer or a user confirmation step before execution in sensitive contexts.
- Log and monitor tool usage; fail closed when validation fails.
from dapr.clients import DaprClient
from dapr.clients.grpc import conversation
@conversation.tool
def get_weather(location: str, unit: str = 'fahrenheit') -> str:
return f"Weather in {location} (unit={unit})"
history: list[conversation.ConversationMessage] = [
conversation.create_user_message("What's the weather in San Francisco?")]
with DaprClient() as client:
# Turn 1
resp1 = client.converse_alpha2(
name="openai",
inputs=[conversation.ConversationInputAlpha2(messages=history)],
tools=conversation.get_registered_tools(),
tool_choice='auto',
temperature=1,
)
# Append assistant messages; execute tool calls; append tool results
for msg in resp1.to_assistant_messages():
history.append(msg)
for tc in msg.of_assistant.tool_calls:
# IMPORTANT: validate inputs and enforce guardrails in production
tool_output = conversation.execute_registered_tool(
tc.function.name, tc.function.arguments
)
history.append(
conversation.create_tool_message(
tool_id=tc.id, name=tc.function.name, content=str(tool_output)
)
)
# Turn 2 (LLM sees tool result)
history.append(conversation.create_user_message("Should I bring an umbrella?"))
resp2 = client.converse_alpha2(
name="openai",
inputs=[conversation.ConversationInputAlpha2(messages=history)],
tools=conversation.get_registered_tools(),
temperature=1,
)
for msg in resp2.to_assistant_messages():
history.append(msg)
if not msg.of_assistant.tool_calls and msg.of_assistant.content:
print(msg.of_assistant.content[0].text)
Tips:
- Always append assistant messages to history.
- Execute each tool call (with validation) and append a tool message with the tool output.
- The next turn includes these tool results so the LLM can reason with them.
When decorators aren’t practical, two options exist.
A) Automatic schema from a typed function:
from enum import Enum
from dapr.clients.grpc import conversation
class Units(Enum):
CELSIUS = 'celsius'
FAHRENHEIT = 'fahrenheit'
def get_weather(location: str, unit: Units = Units.FAHRENHEIT) -> str:
return f"Weather in {location}"
fn = conversation.ConversationToolsFunction.from_function(get_weather)
weather_tool = conversation.ConversationTools(function=fn)
B) Manual JSON Schema (fallback):
from dapr.clients.grpc import conversation
fn = conversation.ConversationToolsFunction(
name='get_weather',
description='Get current weather',
parameters={
'type': 'object',
'properties': {
'location': {'type': 'string'},
'unit': {'type': 'string', 'enum': ['celsius', 'fahrenheit']},
},
'required': ['location'],
},
)
weather_tool = conversation.ConversationTools(function=fn)
Async variant
Use the asynchronous client and async tool execution helpers as needed.
import asyncio
from dapr.aio.clients import DaprClient as AsyncDaprClient
from dapr.clients.grpc import conversation
@conversation.tool
def get_time() -> str:
return '2025-01-01T12:00:00Z'
async def main():
async with AsyncDaprClient() as client:
msg = conversation.create_user_message('What time is it?')
inp = conversation.ConversationInputAlpha2(messages=[msg])
resp = await client.converse_alpha2(
name='openai', inputs=[inp], tools=conversation.get_registered_tools()
)
for m in resp.to_assistant_messages():
if m.of_assistant.content:
print(m.of_assistant.content[0].text)
asyncio.run(main())
If you need to execute tools asynchronously (e.g., network I/O), implement async functions and use conversation.execute_registered_tool_async with timeouts.
Safety and validation (must‑read)
An LLM may suggest tool calls. Treat all model‑provided parameters as untrusted input.
Recommendations:
- Register only trusted functions as tools. Prefer the
@conversation.tool decorator for clarity and automatic schema generation. - Use precise type annotations and docstrings. The SDK converts function signatures to JSON schema and binds parameters with type coercion and rejection of unexpected/invalid fields.
- Add guardrails for tools that can cause side effects (filesystem, network, subprocess). Consider allow‑lists, sandboxing, and limits.
- Validate arguments before execution. For example, sanitize file paths or restrict URLs/domains.
- Consider timeouts and concurrency controls. For async tools, pass a timeout to
execute_registered_tool_async(..., timeout=...). - Log and monitor tool usage. Fail closed: if validation fails, avoid executing the tool and inform the user safely.
See also inline notes in dapr/clients/grpc/conversation.py (e.g., tool(), ConversationTools, execute_registered_tool) for parameter binding and error handling details.
Key helper methods (quick reference)
This section summarizes helper utilities available in dapr.clients.grpc.conversation used throughout the examples.
create_user_message(text: str) -> ConversationMessage
- Builds a user role message for Alpha2. Use in history lists.
- Example:
history.append(conversation.create_user_message("Hello"))
create_system_message(text: str) -> ConversationMessage
- Builds a system message to steer the assistant’s behavior.
- Example:
history = [conversation.create_system_message("You are a concise assistant.")]
create_assistant_message(text: str) -> ConversationMessage
- Useful for injecting assistant text in tests or controlled flows.
create_tool_message(tool_id: str, name: str, content: Any) -> ConversationMessage
- Converts a tool’s output into a tool message the LLM can read next turn.
- content can be any object; it is stringified safely by the SDK.
- Example:
history.append(conversation.create_tool_message(tool_id=tc.id, name=tc.function.name, content=conversation.execute_registered_tool(tc.function.name, tc.function.arguments)))
get_registered_tools() -> list[ConversationTools]
- Returns all tools currently registered in the in-process registry.
- Includes tools created via:
- @conversation.tool decorator (auto-registered by default), and
- ConversationToolsFunction.from_function with register=True (default).
- Pass this list in converse_alpha2(…, tools=…).
register_tool(name: str, t: ConversationTools) / unregister_tool(name: str)
- Manually manage the tool registry (e.g., advanced scenarios, tests, cleanup).
- Names must be unique; unregister to avoid collisions in long-lived processes.
execute_registered_tool(name: str, params: Mapping|Sequence|str|None) -> Any
- Synchronously executes a registered tool by name.
- params accepts kwargs (mapping), args (sequence), JSON string, or None. If a JSON string is provided (as commonly returned by LLMs), it is parsed for you.
- Parameters are validated and coerced against the function signature/schema; unexpected or invalid fields raise errors.
- Security: treat params as untrusted; add guardrails for side effects.
execute_registered_tool_async(name: str, params: Mapping|Sequence|str|None, *, timeout: float|None=None) -> Any
- Async counterpart. Supports timeouts, which are recommended for I/O-bound tools.
- Prefer this for async tools or when using the aio client.
ConversationToolsFunction.from_function(func: Callable, register: bool = True) -> ConversationToolsFunction
- Derives a JSON schema from a typed Python function (annotations + optional docstring) and optionally registers a tool.
- Typical usage:
spec = conversation.ConversationToolsFunction.from_function(my_func); then either rely on auto-registration or wrap with ConversationTools(function=spec) and call register_tool(spec.name, tool) or pass [tool] directly to tools=.
ConversationResponseAlpha2.to_assistant_messages() -> list[ConversationMessage]
- Convenience to transform the response outputs into assistant ConversationMessage objects you can append to history directly (including tool_calls when present).
Tip: The @conversation.tool decorator is the easiest way to create a tool. It auto-generates the schema from your function, allows an optional namespace/name override, and auto-registers the tool (you can set register=False to defer registration).