Conversation API support

README.md

@@ -8,7 +8,9 @@
 [![GitHub issue custom search in repo](https://img.shields.io/github/issues-search/dapr/python-sdk?query=type%3Aissue%20is%3Aopen%20label%3A%22good%20first%20issue%22&label=Good%20first%20issues&style=flat&logo=github)](https://github.com/dapr/python-sdk/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22) 
 [![Discord](https://img.shields.io/discord/778680217417809931?label=Discord&style=flat&logo=discord)](http://bit.ly/dapr-discord) 
 [![YouTube Channel Views](https://img.shields.io/youtube/channel/views/UCtpSQ9BLB_3EXdWAUQYwnRA?style=flat&label=YouTube%20views&logo=youtube)](https://youtube.com/@daprdev) 
+<!-- IGNORE_LINKS -->
 [![X (formerly Twitter) Follow](https://img.shields.io/twitter/follow/daprdev?logo=x&style=flat)](https://twitter.com/daprdev)
+<!-- END_IGNORE -->
 
 [Dapr](https://docs.dapr.io/concepts/overview/) is a portable, event-driven, serverless runtime for building distributed applications across cloud and edge.
 

dapr/aio/clients/grpc/client.py

@@ -29,6 +29,7 @@
 
 from google.protobuf.message import Message as GrpcMessage
 from google.protobuf.empty_pb2 import Empty as GrpcEmpty
+from google.protobuf.any_pb2 import Any as GrpcAny
 
 import grpc.aio  # type: ignore
 from grpc.aio import (  # type: ignore
@@ -75,9 +76,12 @@
     InvokeMethodRequest,
     BindingRequest,
     TransactionalStateOperation,
+    ConversationInput,
 )
 from dapr.clients.grpc._response import (
     BindingResponse,
+    ConversationResponse,
+    ConversationResult,
     DaprResponse,
     GetSecretResponse,
     GetBulkSecretResponse,
@@ -1711,6 +1715,62 @@ async def purge_workflow(self, instance_id: str, workflow_component: str) -> Dap
         except grpc.aio.AioRpcError as err:
             raise DaprInternalError(err.details())
 
+    async def converse_alpha1(
+        self,
+        name: str,
+        inputs: List[ConversationInput],
+        *,
+        context_id: Optional[str] = None,
+        parameters: Optional[Dict[str, GrpcAny]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        scrub_pii: Optional[bool] = None,
+        temperature: Optional[float] = None,
+    ) -> ConversationResponse:
+        """Invoke an LLM using the conversation API (Alpha).
+
+        Args:
+            name: Name of the LLM component to invoke
+            inputs: List of conversation inputs
+            context_id: Optional ID for continuing an existing chat
+            parameters: Optional custom parameters for the request
+            metadata: Optional metadata for the component
+            scrub_pii: Optional flag to scrub PII from inputs and outputs
+            temperature: Optional temperature setting for the LLM to optimize for creativity or predictability
+
+        Returns:
+            ConversationResponse containing the conversation results
+
+        Raises:
+            DaprInternalError: If the Dapr runtime returns an error
+        """
+        inputs_pb = [
+            api_v1.ConversationInput(content=inp.content, role=inp.role, scrubPII=inp.scrub_pii)
+            for inp in inputs
+        ]
+
+        request = api_v1.ConversationRequest(
+            name=name,
+            inputs=inputs_pb,
+            contextID=context_id,
+            parameters=parameters or {},
+            metadata=metadata or {},
+            scrubPII=scrub_pii,
+            temperature=temperature,
+        )
+
+        try:
+            response = await self._stub.ConverseAlpha1(request)
+
+            outputs = [
+                ConversationResult(result=output.result, parameters=output.parameters)
+                for output in response.outputs
+            ]
+
+            return ConversationResponse(context_id=response.contextID, outputs=outputs)
+
+        except Exception as e:
+            raise DaprInternalError(f'Error invoking conversation API: {e}')
+
     async def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

dapr/clients/grpc/_request.py

@@ -15,6 +15,7 @@
 
 import io
 from enum import Enum
+from dataclasses import dataclass
 from typing import Dict, Optional, Union
 
 from google.protobuf.any_pb2 import Any as GrpcAny
@@ -418,3 +419,12 @@ def __next__(self):
 
         self.seq += 1
         return request_proto
+
+
+@dataclass
+class ConversationInput:
+    """A single input message for the conversation."""
+
+    content: str
+    role: Optional[str] = None
+    scrub_pii: Optional[bool] = None

dapr/clients/grpc/_response.py

@@ -18,6 +18,7 @@
 import contextlib
 import json
 import threading
+from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 from typing import (
@@ -1070,3 +1071,19 @@ class EncryptResponse(CryptoResponse[TCryptoResponse]):
 
 class DecryptResponse(CryptoResponse[TCryptoResponse]):
     ...
+
+
+@dataclass
+class ConversationResult:
+    """Result from a single conversation input."""
+
+    result: str
+    parameters: Dict[str, GrpcAny] = field(default_factory=dict)
+
+
+@dataclass
+class ConversationResponse:
+    """Response from the conversation API."""
+
+    context_id: Optional[str]
+    outputs: List[ConversationResult]

dapr/clients/grpc/client.py

@@ -27,6 +27,7 @@
 from datetime import datetime
 from google.protobuf.message import Message as GrpcMessage
 from google.protobuf.empty_pb2 import Empty as GrpcEmpty
+from google.protobuf.any_pb2 import Any as GrpcAny
 
 import grpc  # type: ignore
 from grpc import (  # type: ignore
@@ -64,6 +65,7 @@
     TransactionalStateOperation,
     EncryptRequestIterator,
     DecryptRequestIterator,
+    ConversationInput,
 )
 from dapr.clients.grpc._response import (
     BindingResponse,
@@ -88,6 +90,8 @@
     EncryptResponse,
     DecryptResponse,
     TopicEventResponse,
+    ConversationResponse,
+    ConversationResult,
 )
 
 
@@ -1713,6 +1717,63 @@ def purge_workflow(self, instance_id: str, workflow_component: str) -> DaprRespo
         except RpcError as err:
             raise DaprInternalError(err.details())
 
+    def converse_alpha1(
+        self,
+        name: str,
+        inputs: List[ConversationInput],
+        *,
+        context_id: Optional[str] = None,
+        parameters: Optional[Dict[str, GrpcAny]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        scrub_pii: Optional[bool] = None,
+        temperature: Optional[float] = None,
+    ) -> ConversationResponse:
+        """Invoke an LLM using the conversation API (Alpha).
+
+        Args:
+            name: Name of the LLM component to invoke
+            inputs: List of conversation inputs
+            context_id: Optional ID for continuing an existing chat
+            parameters: Optional custom parameters for the request
+            metadata: Optional metadata for the component
+            scrub_pii: Optional flag to scrub PII from inputs and outputs
+            temperature: Optional temperature setting for the LLM to optimize for creativity or predictability
+
+        Returns:
+            ConversationResponse containing the conversation results
+
+        Raises:
+            DaprInternalError: If the Dapr runtime returns an error
+        """
+
+        inputs_pb = [
+            api_v1.ConversationInput(content=inp.content, role=inp.role, scrubPII=inp.scrub_pii)
+            for inp in inputs
+        ]
+
+        request = api_v1.ConversationRequest(
+            name=name,
+            inputs=inputs_pb,
+            contextID=context_id,
+            parameters=parameters or {},
+            metadata=metadata or {},
+            scrubPII=scrub_pii,
+            temperature=temperature,
+        )
+
+        try:
+            response, call = self.retry_policy.run_rpc(self._stub.ConverseAlpha1.with_call, request)
+
+            outputs = [
+                ConversationResult(result=output.result, parameters=output.parameters)
+                for output in response.outputs
+            ]
+
+            return ConversationResponse(context_id=response.contextID, outputs=outputs)
+
+        except Exception as e:
+            raise DaprInternalError(f'Error invoking conversation API: {e}')
+
     def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.