Python: Fix AI Search Tool Sample and improve AI Search Exceptions (#1206)

* Python: Fix AI Search Tool Sample and improve AI Search Exceptions

* Python: Fix AI Search Tool Test

---------

Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com>

python/samples/getting_started/agents/a2a/agent_with_a2a.py

@@ -53,9 +53,7 @@ async def main():
         resolver = A2ACardResolver(httpx_client=http_client, base_url=a2a_agent_host)
 
         # Get agent card
-        agent_card = await resolver.get_agent_card(
-            relative_card_path="/.well-known/agent.json"
-        )
+        agent_card = await resolver.get_agent_card(relative_card_path="/.well-known/agent.json")
         print(f"Found agent: {agent_card.name} - {agent_card.description}")
 
         # Create A2A agent instance

python/samples/getting_started/agents/azure_ai/azure_ai_with_code_interpreter.py

@@ -43,9 +43,7 @@ async def main() -> None:
     ):
         agent = chat_client.create_agent(
             name="CodingAgent",
-            instructions=(
-                "You are a helpful assistant that can write and execute Python code to solve problems."
-            ),
+            instructions=("You are a helpful assistant that can write and execute Python code to solve problems."),
             tools=HostedCodeInterpreterTool(),
         )
         query = "Generate the factorial of 100 using python code, show the code and execute it."

python/samples/getting_started/agents/azure_ai/azure_ai_with_file_search.py

@@ -63,12 +63,24 @@ async def main() -> None:
                 response = await agent.run(user_input)
                 print(f"# Agent: {response.text}")
 
+            # 5. Cleanup: Delete the vector store and file
+            try:
+                if vector_store:
+                    await client.project_client.agents.vector_stores.delete(vector_store.id)
+                if file:
+                    await client.project_client.agents.files.delete(file.id)
+            except Exception:
+                # Ignore cleanup errors to avoid masking issues
+                pass
     finally:
-        # 5. Cleanup: Delete the vector store and file
+        # 6. Cleanup: Delete the vector store and file in case of eariler failure to prevent orphaned resources.
+
+        # Refreshing the client is required since chat agent closes it
+        client = AzureAIAgentClient(async_credential=AzureCliCredential())
         try:
-            if vector_store is not None:
+            if vector_store:
                 await client.project_client.agents.vector_stores.delete(vector_store.id)
-            if file is not None:
+            if file:
                 await client.project_client.agents.files.delete(file.id)
         except Exception:
             # Ignore cleanup errors to avoid masking issues

python/samples/getting_started/workflows/composition/sub_workflow_parallel_requests.py

@@ -261,11 +261,15 @@ class PolicyEngine(Executor):
 
     @handler
     async def handle_policy_request(
-        self, request: PolicyCheckRequest, ctx: WorkflowContext[RequestResponse[PolicyCheckRequest, Any] | PolicyCheckRequest]
+        self,
+        request: PolicyCheckRequest,
+        ctx: WorkflowContext[RequestResponse[PolicyCheckRequest, Any] | PolicyCheckRequest],
     ) -> None:
         """Handle POLICY requests from sub-workflows and apply rules."""
         policy_request = request
-        print(f"🛡️  POLICY interceptor checking: {policy_request.amount} {policy_request.resource_type}, policy={policy_request.policy_type}")
+        print(
+            f"🛡️  POLICY interceptor checking: {policy_request.amount} {policy_request.resource_type}, policy={policy_request.policy_type}"
+        )
 
         quota_limit = self.quota.get(policy_request.resource_type, 0)
 

python/samples/getting_started/workflows/composition/sub_workflow_request_interception.py

@@ -182,7 +182,7 @@ class SmartEmailOrchestrator(Executor):
     async def handle_domain_request(
         self,
         request: DomainCheckRequest,
-        ctx: WorkflowContext[RequestResponse[DomainCheckRequest, bool] | DomainCheckRequest]
+        ctx: WorkflowContext[RequestResponse[DomainCheckRequest, bool] | DomainCheckRequest],
     ) -> None:
         """Handle requests from sub-workflows."""
         print(f"🔍 Parent intercepting domain check for: {request.domain}")
@@ -190,11 +190,7 @@ class SmartEmailOrchestrator(Executor):
         if request.domain in self.approved_domains:
             print(f"✅ Domain '{request.domain}' is pre-approved locally!")
             # Send response back to sub-workflow
-            response = RequestResponse(
-                data=True,
-                original_request=request,
-                request_id=request.request_id
-            )
+            response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
             await ctx.send_message(response, target_id=request.source_executor_id)
         else:
             print(f"❓ Domain '{request.domain}' unknown, forwarding to external service...")