Python: Fix workflow not pausing when agent calls declaration-only tool (#3757)

* Fix workflow not pausing when agent calls declaration-only tool

* Remove comment
This commit is contained in:
Evan Mattson
2026-02-10 08:51:44 +09:00
committed by GitHub
Unverified
parent e3b4b6662b
commit 6eb251464b
5 changed files with 322 additions and 7 deletions
@@ -84,6 +84,7 @@ Once comfortable with these, explore the rest of the samples below.
| ------------------------------------------ | ------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------- |
| Human-In-The-Loop (Guessing Game) | [human-in-the-loop/guessing_game_with_human_input.py](./human-in-the-loop/guessing_game_with_human_input.py) | Interactive request/response prompts with a human via `ctx.request_info()` |
| Agents with Approval Requests in Workflows | [human-in-the-loop/agents_with_approval_requests.py](./human-in-the-loop/agents_with_approval_requests.py) | Agents that create approval requests during workflow execution and wait for human approval to proceed |
| Agents with Declaration-Only Tools | [human-in-the-loop/agents_with_declaration_only_tools.py](./human-in-the-loop/agents_with_declaration_only_tools.py) | Workflow pauses when agent calls a client-side tool (`func=None`), caller supplies the result |
| SequentialBuilder Request Info | [human-in-the-loop/sequential_request_info.py](./human-in-the-loop/sequential_request_info.py) | Request info for agent responses mid-workflow using `.with_request_info()` on SequentialBuilder |
| ConcurrentBuilder Request Info | [human-in-the-loop/concurrent_request_info.py](./human-in-the-loop/concurrent_request_info.py) | Review concurrent agent outputs before aggregation using `.with_request_info()` on ConcurrentBuilder |
| GroupChatBuilder Request Info | [human-in-the-loop/group_chat_request_info.py](./human-in-the-loop/group_chat_request_info.py) | Steer group discussions with periodic guidance using `.with_request_info()` on GroupChatBuilder |
@@ -0,0 +1,95 @@
# Copyright (c) Microsoft. All rights reserved.
"""
Sample: Declaration-only tools in a workflow (issue #3425)
A declaration-only tool (func=None) represents a client-side tool that the
framework cannot execute — the LLM can call it, but the workflow must pause
so the caller can supply the result.
Flow:
1. The agent is given a declaration-only tool ("get_user_location").
2. When the LLM decides to call it, the workflow pauses and emits a
request_info event containing the FunctionCallContent.
3. The caller inspects the tool name/args, runs the tool however it wants,
and feeds the result back via workflow.run(responses={...}).
4. The workflow resumes — the agent sees the tool result and finishes.
Prerequisites:
- Azure OpenAI endpoint configured via environment variables.
- `az login` for AzureCliCredential.
"""
import asyncio
import json
from typing import Any
from agent_framework import Content, FunctionTool, WorkflowBuilder
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
# A declaration-only tool: the schema is sent to the LLM, but the framework
# has no implementation to execute. The caller must supply the result.
get_user_location = FunctionTool(
name="get_user_location",
func=None,
description="Get the user's current city. Only the client application can resolve this.",
input_model={
"type": "object",
"properties": {
"reason": {"type": "string", "description": "Why the location is needed"},
},
"required": ["reason"],
},
)
async def main() -> None:
agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
name="WeatherBot",
instructions=(
"You are a helpful weather assistant. "
"When the user asks about weather, call get_user_location first, "
"then make up a plausible forecast for that city."
),
tools=[get_user_location],
)
workflow = WorkflowBuilder(start_executor=agent).build()
# --- First run: the agent should call the declaration-only tool ---
print(">>> Sending: 'What's the weather like today?'")
result = await workflow.run("What's the weather like today?")
requests = result.get_request_info_events()
if not requests:
# The LLM chose not to call the tool — print whatever it said and exit
print(f"Agent replied without calling the tool: {result.get_outputs()}")
return
# --- Inspect what the agent wants ---
for req in requests:
data = req.data
args = json.loads(data.arguments) if isinstance(data.arguments, str) else data.arguments
print(f"Workflow paused — agent called: {data.name}({args})")
# --- "Execute" the tool on the client side and send results back ---
responses: dict[str, Any] = {}
for req in requests:
# In a real app this could be a GPS lookup, browser API, user prompt, etc.
client_result = "Seattle, WA"
print(f"Client provides result for {req.data.name}: {client_result!r}")
responses[req.request_id] = Content.from_function_result(
call_id=req.data.call_id,
result=client_result,
)
result = await workflow.run(responses=responses)
# --- Final answer ---
for output in result.get_outputs():
print(f"\nAgent: {output.text}")
if __name__ == "__main__":
asyncio.run(main())