Implementation:Openai Openai agents python MCP Approval Callback Pattern
| Knowledge Sources | |
|---|---|
| Domains | MCP Integration, Human-in-the-Loop, Tool Approval |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Demonstrates callback-based MCP tool approval using the on_approval_request parameter on HostedMCPTool, enabling human-in-the-loop confirmation before MCP tools are executed.
Description
The MCP Approval Callback pattern provides a mechanism for requiring user approval before MCP tool invocations are executed. When require_approval is set to "always" in the tool configuration, every MCP tool call triggers an approval request that is routed to the callback function specified in on_approval_request.
The callback function receives an MCPToolApprovalRequest object containing the tool name (request.data.name) and its arguments (request.data.arguments). It must return an MCPToolApprovalFunctionResult dictionary with an "approve" boolean field and an optional "reason" string explaining the rejection. In this example, the callback prompts the user interactively via the terminal, displaying the tool name and parameters before asking for confirmation.
The example uses a DeepWiki hosted MCP server to answer questions about GitHub repositories. In non-streaming mode, the run may return interruptions when tools require approval; the loop pattern while run_result.interruptions: run_result = await Runner.run(agent, run_result.to_state()) handles re-submitting the run after approvals are processed. In streaming mode, approval requests are handled inline during event iteration.
Usage
Use this pattern when MCP tool invocations have side effects that require human oversight, such as modifying external data, executing commands, or accessing sensitive resources. It is essential for production deployments where unattended tool execution poses security or compliance risks.
Code Reference
Source Location
- Repository: Openai_Openai_agents_python
- File: examples/hosted_mcp/on_approval.py
- Lines: 1-85
Signature
HostedMCPTool(
tool_config={
"type": "mcp",
"server_label": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "always",
},
on_approval_request=prompt_approval,
)
Import
from agents import (
Agent,
HostedMCPTool,
MCPToolApprovalFunctionResult,
MCPToolApprovalRequest,
Runner,
RunResult,
RunResultStreaming,
)
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| tool_config.type | str |
Yes | Must be "mcp"
|
| tool_config.server_label | str |
Yes | Human-readable label for the MCP server |
| tool_config.server_url | str |
Yes | URL of the hosted MCP server endpoint |
| tool_config.require_approval | Literal["always"] |
Yes | Set to "always" to require approval for every tool call
|
| on_approval_request | Callable[[MCPToolApprovalRequest], MCPToolApprovalFunctionResult] |
Yes | Callback function that receives approval requests and returns approval decisions |
| request.data.name | str |
N/A | The name of the MCP tool being called (received in callback) |
| request.data.arguments | None | N/A | The arguments passed to the MCP tool (received in callback) |
Outputs
| Name | Type | Description |
|---|---|---|
| MCPToolApprovalFunctionResult["approve"] | bool |
Whether the tool call is approved |
| MCPToolApprovalFunctionResult["reason"] | None | Optional reason for rejection |
| run_result.final_output | str |
The agent's final text response |
| run_result.interruptions | None | Pending interruptions requiring approval (non-streaming mode) |
Usage Examples
Human-in-the-Loop Approval with Non-Streaming Run
import asyncio
import json
from agents import (
Agent,
HostedMCPTool,
MCPToolApprovalFunctionResult,
MCPToolApprovalRequest,
Runner,
)
def prompt_approval(request: MCPToolApprovalRequest) -> MCPToolApprovalFunctionResult:
params = request.data.arguments or {}
user_input = input(
f"Approve tool '{request.data.name}' with params {json.dumps(params)}? (y/n) "
)
approved = user_input.strip().lower() == "y"
result: MCPToolApprovalFunctionResult = {"approve": approved}
if not approved:
result["reason"] = "User denied"
return result
async def main():
agent = Agent(
name="MCP Assistant",
instructions="Use MCP tools to answer questions.",
tools=[
HostedMCPTool(
tool_config={
"type": "mcp",
"server_label": "deepwiki",
"server_url": "https://mcp.deepwiki.com/mcp",
"require_approval": "always",
},
on_approval_request=prompt_approval,
)
],
)
run_result = await Runner.run(agent, "Which language is openai/codex written in?")
while run_result.interruptions:
run_result = await Runner.run(agent, run_result.to_state())
print(run_result.final_output)
asyncio.run(main())