Add RunHandler to support MCP toolcall approval and manual function tool call for runs.create_and_process (#42855)

* Add new event handler for create_and_process

* Resolved comments

Author: howieleung

sdk/ai/azure-ai-agents/CHANGELOG.md

@@ -5,11 +5,16 @@
 ## 1.2.0b5 (Unreleased)
 
 ### Features Added
+- Added `run_handler` parameter to `runs.create_and_process` allowing to make function tool calls manually or approve mcp tool calls.
 
 ### Bugs Fixed
-- Added `RunStepDeltaChunk` to `StreamEventData` model (GitHub issues [43022](https://github.com/Azure/azure-sdk-for-python/issues/43022))
 - Fixed regression, reverted ToolOutput type signature and usage in tool_output submission.  
 - Added `RunStepDeltaComputerUseDetails` and `RunStepDeltaComputerUseToolCall` classes for streaming computer use scenarios.
+- Added `RunStepDeltaChunk` to `StreamEventData` model (GitHub issues [43022](https://github.com/Azure/azure-sdk-for-python/issues/43022))
+
+### Sample updates
+- Added `sample_agents_mcp_in_create_and_process.py` abd `sample_agents_mcp_in_create_and_process_async.py` demonstrating MCP tool call approvals in `runs.create_and_process`.
+- Added `sample_agents_functions_in_create_and_process.py` and `sample_agents_functions_in_create_and_process_async.py` demonstrating manual function tool calls in `runs.create_and_process`.
 
 ## 1.2.0b4 (2025-09-12)
 

sdk/ai/azure-ai-agents/README.md

@@ -514,6 +514,38 @@ while run.status in ["queued", "in_progress", "requires_action"]:
 
 <!-- END SNIPPET -->
 
+For scenarios requiring custom approval logic or additional control over MCP tool calls, you can use a `RunHandler` with the `create_and_process` method. This approach allows you to implement custom logic for approving or denying MCP tool calls.
+
+Here's an example that demonstrates manual MCP tool call approval using a `RunHandler`:
+
+<!-- SNIPPET:sample_agents_mcp_in_create_and_process.run_handler -->
+
+```python
+class MyRunHandler(RunHandler):
+    def submit_mcp_tool_approval(
+        self, *, run: ThreadRun, tool_call: RequiredMcpToolCall, **kwargs: Any
+    ) -> ToolApproval:
+        return ToolApproval(
+            tool_call_id=tool_call.id,
+            approve=True,
+            headers=mcp_tool.headers,
+        )
+```
+
+<!-- END SNIPPET -->
+
+To use the RunHandler with `create_and_process` for MCP tools:
+
+<!-- SNIPPET:sample_agents_mcp_in_create_and_process.create_and_process -->
+
+```python
+run = agents_client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id, run_handler=MyRunHandler())
+```
+
+<!-- END SNIPPET -->
+
+For a complete example, see [`sample_agents_mcp_in_create_and_process.py`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_tools/sample_agents_mcp_in_create_and_process.py).
+
 ### Create Agent with Azure AI Search
 
 Azure AI Search is an enterprise search system for high-performance applications. It integrates with Azure OpenAI Service and Azure Machine Learning, offering advanced search technologies like vector search and full-text search. Ideal for knowledge base insights, information discovery, and automation. Creating an Agent with Azure AI Search requires an existing Azure AI Search Index. For more information and setup guides, see [Azure AI Search Tool Guide](https://learn.microsoft.com/azure/ai-services/agents/how-to/tools/azure-ai-search?tabs=azurecli%2Cpython&pivots=overview-azure-ai-search).
@@ -634,7 +666,6 @@ When `enable_auto_function_calls` is called, the SDK will automatically invoke f
 - If you prefer to manage function execution manually, refer to [`sample_agents_stream_eventhandler_with_functions.py`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_streaming/sample_agents_stream_eventhandler_with_functions.py) or
 [`sample_agents_functions.py`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_tools/sample_agents_functions.py).
 
-
 ### Create Agent With Azure Function Call
 
 The AI agent leverages Azure Functions triggered asynchronously via Azure Storage Queues. To enable the agent to perform Azure Function calls, you must set up the corresponding `AzureFunctionTool`, specifying input and output queues as well as parameter definitions.
@@ -1366,6 +1397,44 @@ run = agents_client.runs.create_and_process(thread_id=thread.id, agent_id=agent.
 
 <!-- END SNIPPET -->
 
+For scenarios requiring manual approval or custom control over function execution (such as security-sensitive operations), you can use a `RunHandler` with the `create_and_process` method. This approach allows you to implement custom logic to decide whether, when, and how functions should be executed.
+
+Here's an example that demonstrates manual function calls using a `RunHandler`:
+
+<!-- SNIPPET:sample_agents_functions_in_create_and_process.run_handler -->
+
+```python
+class MyRunHandler(RunHandler):
+    def submit_function_call_output(
+        self,
+        *,
+        run: ThreadRun,
+        tool_call: RequiredFunctionToolCall,
+        tool_call_details: RequiredFunctionToolCallDetails,
+        **kwargs: Any,
+    ) -> Any:
+        function_name = tool_call_details.name
+        if function_name == send_email.__name__:
+            # Parse arguments from tool call
+            args_dict = json.loads(tool_call_details.arguments) if tool_call_details.arguments else {}
+            # Call the function directly with the arguments
+            return send_email(**args_dict)
+```
+
+<!-- END SNIPPET -->
+
+To use the RunHandler with `create_and_process`:
+
+<!-- SNIPPET:sample_agents_functions_in_create_and_process.create_and_process -->
+
+```python
+run = agents_client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id, run_handler=MyRunHandler())
+```
+
+<!-- END SNIPPET -->
+
+For a complete example, see [`sample_agents_functions_in_create_and_process.py`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_tools/sample_agents_functions_in_create_and_process.py).
+
 With streaming, polling need not be considered. If `function tools` were added to the agents, you should decide to have the function tools called manually or automatically.  Please visit [`manual function call sample`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_streaming/sample_agents_stream_eventhandler_with_functions.py) or [`automatic function call sample`](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agents/samples/agents_streaming/sample_agents_stream_iteration_with_toolset.py).    
 
 Here is a basic example of streaming:

sdk/ai/azure-ai-agents/azure/ai/agents/aio/operations/_patch.py

@@ -32,7 +32,7 @@
 from azure.core.tracing.decorator_async import distributed_trace_async
 
 from ... import models as _models
-from ...models._enums import FilePurpose, RunStatus
+from ...models._enums import FilePurpose
 from ._operations import FilesOperations as FilesOperationsGenerated
 from ._operations import MessagesOperations as MessagesOperationsGenerated
 from ._operations import RunsOperations as RunsOperationsGenerated
@@ -442,6 +442,7 @@ async def create_and_process(
         response_format: Optional["_types.AgentsResponseFormatOption"] = None,
         parallel_tool_calls: Optional[bool] = None,
         metadata: Optional[Dict[str, str]] = None,
+        run_handler: Optional[_models.AsyncRunHandler] = None,
         polling_interval: int = 1,
         **kwargs: Any,
     ) -> _models.ThreadRun:
@@ -522,6 +523,9 @@ async def create_and_process(
          64 characters in length and values may be up to 512 characters in length. Default value is
          None.
         :paramtype metadata: dict[str, str]
+        :keyword run_handler: Optional handler to customize run processing and tool execution.
+            Default value is None.
+        :paramtype run_handler: ~azure.ai.agents.models.AsyncRunHandler
         :keyword polling_interval: The time in seconds to wait between polling the service for run status.
             Default value is 1.
         :paramtype polling_interval: int
@@ -553,51 +557,9 @@ async def create_and_process(
         )
 
         # Monitor and process the run status
-        current_retry = 0
-        while run.status in [
-            RunStatus.QUEUED,
-            RunStatus.IN_PROGRESS,
-            RunStatus.REQUIRES_ACTION,
-        ]:
-            await asyncio.sleep(polling_interval)
-            run = await self.get(thread_id=thread_id, run_id=run.id)
-
-            if run.status == "requires_action" and isinstance(run.required_action, _models.SubmitToolOutputsAction):
-                tool_calls = run.required_action.submit_tool_outputs.tool_calls
-                if not tool_calls:
-                    logger.warning("No tool calls provided - cancelling run")
-                    await self.cancel(thread_id=thread_id, run_id=run.id)
-                    break
-                # We need tool set only if we are executing local function. In case if
-                # the tool is azure_function we just need to wait when it will be finished.
-                if any(tool_call.type == "function" for tool_call in tool_calls):
-                    toolset = _models.AsyncToolSet()
-                    toolset.add(self._function_tool)
-                    tool_outputs = await toolset.execute_tool_calls(tool_calls)
-
-                    if _has_errors_in_toolcalls_output(tool_outputs):
-                        if current_retry >= self._function_tool_max_retry:  # pylint:disable=no-else-return
-                            logger.warning(
-                                "Tool outputs contain errors - reaching max retry %s", self._function_tool_max_retry
-                            )
-                            return await self.cancel(thread_id=thread_id, run_id=run.id)
-                        else:
-                            logger.warning("Tool outputs contain errors - retrying")
-                            current_retry += 1
-
-                    logger.debug("Tool outputs: %s", tool_outputs)
-                    if tool_outputs:
-                        run2 = await self.submit_tool_outputs(
-                            thread_id=thread_id, run_id=run.id, tool_outputs=tool_outputs
-                        )
-                        logger.debug("Tool outputs submitted to run: %s", run2.id)
-            elif isinstance(run.required_action, _models.SubmitToolApprovalAction):
-                logger.warning("Automatic MCP tool approval is not supported.")
-                await self.cancel(thread_id=thread_id, run_id=run.id)
-
-            logger.debug("Current run ID: %s with status: %s", run.id, run.status)
-
-        return run
+        run_handler_obj = run_handler or _models.AsyncRunHandler()
+
+        return await run_handler_obj._start(self, run, polling_interval)  # pylint: disable=protected-access
 
     @overload
     async def stream(