Adds async function calling.

Async function calls are non-blocking function calls that are particularly well suited for realtime agents. They allow long-running operations to be run on the background without interrupting the main realtime agents. When such functions return a stream of parts and are implemented with processors, async function calls provide a simple mechanism to implement multi-agents.

PiperOrigin-RevId: 828839605

Author: aelissee

genai_processors/content_api.py

@@ -332,6 +332,7 @@ def from_function_response(
       function_call_id: str | None = None,
       will_continue: bool | None = None,
       scheduling: genai_types.FunctionResponseScheduling | None = None,
+      is_error: bool = False,
       **kwargs,
   ) -> 'ProcessorPart':
     """Constructs a ProcessorPart as a function response.
@@ -349,12 +350,19 @@ def from_function_response(
       scheduling: The scheduling policy for the function response. Controls
         whether the response generation will be triggered immediately or the
         function response is just added to the context.
+      is_error: Whether the function response is an error. When it is an error,
+        the response will be stored in the '.error' field in the function
+        response.
       **kwargs: Additional keyword arguments to pass to the `ProcessorPart`
         constructor.
 
     Returns:
       A ProcessorPart representing the function response.
     """
+    if is_error:
+      response_key = 'error'
+    else:
+      response_key = 'result'
     function_response_args = dict(
         id=function_call_id,
         name=name,
@@ -365,15 +373,15 @@ def from_function_response(
     # serializable object. This is what tools in Gemini API return historically.
     try:
       function_response = genai_types.FunctionResponse(
-          response={'result': response}, **function_response_args
+          response={response_key: response}, **function_response_args
       )
       function_response.json()
     except ValueError:
       # Response is not JSON serializable. Then try to construct content.
       response_content = ProcessorContent(response)
       try:
         function_response = genai_types.FunctionResponse(
-            response={'result': as_text(response_content, strict=True)},
+            response={response_key: as_text(response_content, strict=True)},
             **function_response_args,
         )
       except ValueError:
@@ -445,12 +453,16 @@ def from_dataclass(cls, *, dataclass: Any, **kwargs) -> 'ProcessorPart':
     return cls(part, **kwargs)
 
   @classmethod
-  def end_of_turn(cls, substream_name: str = '') -> 'ProcessorPart':
+  def end_of_turn(
+      cls, substream_name: str = '', metadata: dict[str, Any] | None = None
+  ) -> 'ProcessorPart':
+    metadata = metadata or {}
+    metadata['turn_complete'] = True
     return ProcessorPart(
         '',
         role='user',
         substream_name=substream_name,
-        metadata={'turn_complete': True},
+        metadata=metadata,
     )
 
   @classmethod