Python: Replaced obsolete create_response method in samples (#3542)

* Replaced obsolete create_response method in samples

* Addressed PR comments
This commit is contained in:
Dmytro Struk
2026-01-30 14:22:00 -08:00
committed by GitHub
Unverified
parent f3e0be9555
commit 493891620d
12 changed files with 46 additions and 48 deletions
@@ -6,8 +6,7 @@ from typing import Annotated
from agent_framework import (
ChatMessage,
ConcurrentBuilder,
FunctionApprovalRequestContent,
FunctionApprovalResponseContent,
Content,
RequestInfoEvent,
WorkflowOutputEvent,
tool,
@@ -139,7 +138,7 @@ async def main() -> None:
):
if isinstance(event, RequestInfoEvent):
request_info_events.append(event)
if isinstance(event.data, FunctionApprovalRequestContent):
if isinstance(event.data, Content) and event.data.type == "function_approval_request":
print(f"\nApproval requested for tool: {event.data.function_call.name}")
print(f" Arguments: {event.data.function_call.arguments}")
elif isinstance(event, WorkflowOutputEvent):
@@ -147,12 +146,12 @@ async def main() -> None:
# 6. Handle approval requests (if any)
if request_info_events:
responses: dict[str, FunctionApprovalResponseContent] = {}
responses: dict[str, Content] = {}
for request_event in request_info_events:
if isinstance(request_event.data, FunctionApprovalRequestContent):
if isinstance(request_event.data, Content) and request_event.data.type == "function_approval_request":
print(f"\nSimulating human approval for: {request_event.data.function_call.name}")
# Create approval response
responses[request_event.request_id] = request_event.data.create_response(approved=True)
responses[request_event.request_id] = request_event.data.to_function_approval_response(approved=True)
if responses:
# Phase 2: Send all approvals and continue workflow
@@ -5,7 +5,7 @@ from typing import Annotated
from agent_framework import (
AgentRunUpdateEvent,
FunctionApprovalRequestContent,
Content,
GroupChatBuilder,
GroupChatRequestSentEvent,
GroupChatState,
@@ -144,7 +144,7 @@ async def main() -> None:
):
if isinstance(event, RequestInfoEvent):
request_info_events.append(event)
if isinstance(event.data, FunctionApprovalRequestContent):
if isinstance(event.data, Content) and event.data.type == "function_approval_request":
print("\n[APPROVAL REQUIRED] From agent:", event.source_executor_id)
print(f" Tool: {event.data.function_call.name}")
print(f" Arguments: {event.data.function_call.arguments}")
@@ -164,7 +164,7 @@ async def main() -> None:
# 6. Handle approval requests
if request_info_events:
for request_event in request_info_events:
if isinstance(request_event.data, FunctionApprovalRequestContent):
if isinstance(request_event.data, Content) and request_event.data.type == "function_approval_request":
print("\n" + "=" * 60)
print("Human review required for production deployment!")
print("In a real scenario, you would review the deployment details here.")
@@ -172,7 +172,7 @@ async def main() -> None:
print("=" * 60)
# Create approval response
approval_response = request_event.data.create_response(approved=True)
approval_response = request_event.data.to_function_approval_response(approved=True)
# Phase 2: Send approval and continue workflow
# Keep track of the response to format output nicely in streaming mode
@@ -5,7 +5,7 @@ from typing import Annotated
from agent_framework import (
ChatMessage,
FunctionApprovalRequestContent,
Content,
RequestInfoEvent,
SequentialBuilder,
WorkflowOutputEvent,
@@ -23,7 +23,7 @@ with approval_mode="always_require" to trigger human-in-the-loop interactions.
This sample works as follows:
1. A SequentialBuilder workflow is created with a single agent that has tools requiring approval.
2. The agent receives a user task and determines it needs to call a sensitive tool.
3. The tool call triggers a FunctionApprovalRequestContent, pausing the workflow.
3. The tool call triggers a function_approval_request Content, pausing the workflow.
4. The sample simulates human approval by responding to the RequestInfoEvent.
5. Once approved, the tool executes and the agent completes its response.
6. The workflow outputs the final conversation with all messages.
@@ -34,7 +34,7 @@ requiring any additional builder configuration.
Demonstrate:
- Using @tool(approval_mode="always_require") for sensitive operations.
- Handling RequestInfoEvent with FunctionApprovalRequestContent in sequential workflows.
- Handling RequestInfoEvent with function_approval_request Content in sequential workflows.
- Resuming workflow execution after approval via send_responses_streaming.
Prerequisites:
@@ -92,19 +92,19 @@ async def main() -> None:
):
if isinstance(event, RequestInfoEvent):
request_info_events.append(event)
if isinstance(event.data, FunctionApprovalRequestContent):
if isinstance(event.data, Content) and event.data.type == "function_approval_request":
print(f"\nApproval requested for tool: {event.data.function_call.name}")
print(f" Arguments: {event.data.function_call.arguments}")
# 5. Handle approval requests
if request_info_events:
for request_event in request_info_events:
if isinstance(request_event.data, FunctionApprovalRequestContent):
if isinstance(request_event.data, Content) and request_event.data.type == "function_approval_request":
# In a real application, you would prompt the user here
print("\nSimulating human approval (auto-approving for demo)...")
# Create approval response
approval_response = request_event.data.create_response(approved=True)
approval_response = request_event.data.to_function_approval_response(approved=True)
# Phase 2: Send approval and continue workflow
output: list[ChatMessage] | None = None