Repository navigation
Expand file tree
/
Copy pathfunction_tool_with_approval_and_sessions.py
More file actions
109 lines (80 loc) · 3.56 KB
/
Copy pathfunction_tool_with_approval_and_sessions.py
File metadata and controls
109 lines (80 loc) · 3.56 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from typing import Annotated
from agent_framework import Agent, Message, tool
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
"""
Tool Approvals with Sessions
This sample demonstrates using tool approvals with sessions.
With sessions, you don't need to manually pass previous messages -
the session stores and retrieves them automatically.
"""
@tool(approval_mode="always_require")
def add_to_calendar(event_name: Annotated[str, "Name of the event"], date: Annotated[str, "Date of the event"]) -> str:
"""Add an event to the calendar (requires approval)."""
print(f">>> EXECUTING: add_to_calendar(event_name='{event_name}', date='{date}')")
return f"Added '{event_name}' to calendar on {date}"
async def approval_example() -> None:
"""Example showing approval with sessions."""
print("=== Tool Approval with Session ===\n")
agent = Agent(
client=FoundryChatClient(credential=AzureCliCredential()),
name="CalendarAgent",
instructions="You are a helpful calendar assistant.",
tools=[add_to_calendar],
)
session = agent.create_session()
# Step 1: Agent requests to call the tool
query = "Add a dentist appointment on March 15th"
print(f"User: {query}")
result = await agent.run(query, session=session)
# Check for approval requests
if result.user_input_requests:
for request in result.user_input_requests:
if request.function_call is None:
continue
print("\nApproval needed:")
print(f" Function: {request.function_call.name}")
print(f" Arguments: {request.function_call.arguments}")
# User approves (in real app, this would be user input)
approved = True # Change to False to see rejection
print(f" Decision: {'Approved' if approved else 'Rejected'}")
# Step 2: Send approval response
approval_response = request.to_function_approval_response(approved=approved)
result = await agent.run(Message("user", [approval_response]), session=session)
print(f"Agent: {result}\n")
async def rejection_example() -> None:
"""Example showing rejection with sessions."""
print("=== Tool Rejection with Session ===\n")
agent = Agent(
client=FoundryChatClient(credential=AzureCliCredential()),
name="CalendarAgent",
instructions="You are a helpful calendar assistant.",
tools=[add_to_calendar],
)
session = agent.create_session()
query = "Add a team meeting on December 20th"
print(f"User: {query}")
result = await agent.run(query, session=session)
if result.user_input_requests:
for request in result.user_input_requests:
if request.function_call is None:
continue
print("\nApproval needed:")
print(f" Function: {request.function_call.name}")
print(f" Arguments: {request.function_call.arguments}")
# User rejects
print(" Decision: Rejected")
# Send rejection response
rejection_response = request.to_function_approval_response(approved=False)
result = await agent.run(Message("user", [rejection_response]), session=session)
print(f"Agent: {result}\n")
async def main() -> None:
await approval_example()
await rejection_example()
if __name__ == "__main__":
asyncio.run(main())