mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Added Shell tool (#4339)
* Added shell tool * Fixed CI error * Add ShellTool support for OpenAI and Anthropic providers - Add shell_tool_call, shell_tool_result, and shell_command_output content types - Add ShellTool class and shell_tool decorator to core - Add get_hosted_shell_tool() to OpenAI Responses client - Handle shell_call and shell_call_output parsing in OpenAI (sync and streaming) - Map ShellTool to Anthropic bash tool API format - Parse bash_code_execution_tool_result as shell_tool_result in Anthropic - Add unit tests for all new functionality - Add sample scripts for hosted and local shell execution Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Addressed comments * Reverted ruff change * Fixed tests * Addressed comments --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Dmytro Struk
committed by
GitHub
Unverified
GitHub
parent
dae3caa719
commit
1c0ae4b659
@@ -0,0 +1,100 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import subprocess
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import Agent, Message, tool
|
||||
from agent_framework.anthropic import AnthropicClient
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables from .env file
|
||||
load_dotenv()
|
||||
|
||||
"""
|
||||
Anthropic Client with Shell Tool Example
|
||||
|
||||
This sample demonstrates using @tool(approval_mode=...) with AnthropicClient
|
||||
for executing bash commands locally. The bash tool tells the model it can
|
||||
request shell commands, while the actual execution happens on YOUR machine
|
||||
via a user-provided function.
|
||||
|
||||
SECURITY NOTE: This example executes real commands on your local machine.
|
||||
Only enable this when you trust the agent's actions. Consider implementing
|
||||
allowlists, sandboxing, or approval workflows for production use.
|
||||
"""
|
||||
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def run_bash(command: str) -> str:
|
||||
"""Execute a bash command using subprocess and return the output."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
parts: list[str] = []
|
||||
if result.stdout:
|
||||
parts.append(result.stdout)
|
||||
if result.stderr:
|
||||
parts.append(f"stderr: {result.stderr}")
|
||||
parts.append(f"exit_code: {result.returncode}")
|
||||
return "\n".join(parts)
|
||||
except subprocess.TimeoutExpired:
|
||||
return "Command timed out after 30 seconds"
|
||||
except Exception as e:
|
||||
return f"Error executing command: {e}"
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
"""Example showing how to use the shell tool with AnthropicClient."""
|
||||
print("=== Anthropic Agent with Shell Tool Example ===")
|
||||
print("NOTE: Commands will execute on your local machine.\n")
|
||||
|
||||
client = AnthropicClient()
|
||||
shell = client.get_shell_tool(func=run_bash)
|
||||
agent = Agent(
|
||||
client=client,
|
||||
instructions="You are a helpful assistant that can execute bash commands to answer questions.",
|
||||
tools=[shell],
|
||||
)
|
||||
|
||||
query = "Use bash to print 'Hello from Anthropic shell!' and show the current working directory"
|
||||
print(f"User: {query}")
|
||||
result = await run_with_approvals(query, agent)
|
||||
print(f"Result: {result}\n")
|
||||
|
||||
|
||||
async def run_with_approvals(query: str, agent: Agent) -> Any:
|
||||
"""Run the agent and handle shell approvals outside tool execution."""
|
||||
current_input: str | list[Any] = query
|
||||
while True:
|
||||
result = await agent.run(current_input)
|
||||
if not result.user_input_requests:
|
||||
return result
|
||||
|
||||
next_input: list[Any] = [query]
|
||||
rejected = False
|
||||
for user_input_needed in result.user_input_requests:
|
||||
print(
|
||||
f"\nShell request: {user_input_needed.function_call.name}"
|
||||
f"\nArguments: {user_input_needed.function_call.arguments}"
|
||||
)
|
||||
user_approval = await asyncio.to_thread(input, "\nApprove shell command? (y/n): ")
|
||||
approved = user_approval.strip().lower() == "y"
|
||||
next_input.append(Message("assistant", [user_input_needed]))
|
||||
next_input.append(Message("user", [user_input_needed.to_function_approval_response(approved)]))
|
||||
if not approved:
|
||||
rejected = True
|
||||
break
|
||||
if rejected:
|
||||
print("\nShell command rejected. Stopping without additional approval prompts.")
|
||||
return "Shell command execution was rejected by user."
|
||||
current_input = next_input
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+116
@@ -0,0 +1,116 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import subprocess
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import Agent, Message, tool
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables from .env file
|
||||
load_dotenv()
|
||||
|
||||
"""
|
||||
OpenAI Responses Client with Local Shell Tool Example
|
||||
|
||||
This sample demonstrates implementing a local shell tool using get_shell_tool(func=...)
|
||||
that wraps Python's subprocess module. Unlike the hosted shell tool (get_shell_tool()),
|
||||
local shell execution runs commands on YOUR machine, not in a remote container.
|
||||
|
||||
SECURITY NOTE: This example executes real commands on your local machine.
|
||||
Only enable this when you trust the agent's actions. Consider implementing
|
||||
allowlists, sandboxing, or approval workflows for production use.
|
||||
"""
|
||||
|
||||
|
||||
@tool(approval_mode="always_require")
|
||||
def run_bash(command: str) -> str:
|
||||
"""Execute a shell command locally and return stdout, stderr, and exit code."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
parts: list[str] = []
|
||||
if result.stdout:
|
||||
parts.append(result.stdout)
|
||||
if result.stderr:
|
||||
parts.append(f"stderr: {result.stderr}")
|
||||
parts.append(f"exit_code: {result.returncode}")
|
||||
return "\n".join(parts)
|
||||
except subprocess.TimeoutExpired:
|
||||
return "Command timed out after 30 seconds"
|
||||
except Exception as e:
|
||||
return f"Error executing command: {e}"
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
"""Example showing how to use a local shell tool with OpenAI."""
|
||||
print("=== OpenAI Agent with Local Shell Tool Example ===")
|
||||
print("NOTE: Commands will execute on your local machine.\n")
|
||||
|
||||
client = OpenAIResponsesClient()
|
||||
local_shell_tool = client.get_shell_tool(
|
||||
func=run_bash,
|
||||
)
|
||||
|
||||
agent = Agent(
|
||||
client=client,
|
||||
instructions="You are a helpful assistant that can run shell commands to help the user.",
|
||||
tools=[local_shell_tool],
|
||||
)
|
||||
|
||||
query = "Use the run_bash tool to execute `python --version` and show only the command output."
|
||||
print(f"User: {query}")
|
||||
result = await run_with_approvals(query, agent)
|
||||
if isinstance(result, str):
|
||||
print(f"Agent: {result}\n")
|
||||
return
|
||||
if result.text:
|
||||
print(f"Agent: {result.text}\n")
|
||||
else:
|
||||
printed = False
|
||||
for message in result.messages:
|
||||
for content in message.contents:
|
||||
if content.type == "function_result" and content.result:
|
||||
print(f"Agent (tool output): {content.result}\n")
|
||||
printed = True
|
||||
if not printed:
|
||||
print("Agent: (no text output returned)\n")
|
||||
|
||||
|
||||
async def run_with_approvals(query: str, agent: Agent) -> Any:
|
||||
"""Run the agent and handle shell approvals outside tool execution."""
|
||||
current_input: str | list[Any] = query
|
||||
|
||||
while True:
|
||||
result = await agent.run(current_input)
|
||||
if not result.user_input_requests:
|
||||
return result
|
||||
|
||||
next_input: list[Any] = [query]
|
||||
rejected = False
|
||||
for user_input_needed in result.user_input_requests:
|
||||
print(
|
||||
f"\nShell request: {user_input_needed.function_call.name}"
|
||||
f"\nArguments: {user_input_needed.function_call.arguments}"
|
||||
)
|
||||
user_approval = await asyncio.to_thread(input, "\nApprove shell command? (y/n): ")
|
||||
approved = user_approval.strip().lower() == "y"
|
||||
next_input.append(Message("assistant", [user_input_needed]))
|
||||
next_input.append(Message("user", [user_input_needed.to_function_approval_response(approved)]))
|
||||
if not approved:
|
||||
rejected = True
|
||||
break
|
||||
if rejected:
|
||||
print("\nShell command rejected. Stopping without additional approval prompts.")
|
||||
return "Shell command execution was rejected by user."
|
||||
current_input = next_input
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,61 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
|
||||
from agent_framework import Agent
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables from .env file
|
||||
load_dotenv()
|
||||
|
||||
"""
|
||||
OpenAI Responses Client with Shell Tool Example
|
||||
|
||||
This sample demonstrates using get_shell_tool() with OpenAI Responses Client
|
||||
for executing shell commands in a managed container environment hosted by OpenAI.
|
||||
|
||||
The shell tool allows the model to run commands like listing files, running scripts,
|
||||
or performing system operations within a secure, sandboxed container.
|
||||
"""
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
"""Example showing how to use the shell tool with OpenAI Responses."""
|
||||
print("=== OpenAI Responses Agent with Shell Tool Example ===")
|
||||
|
||||
client = OpenAIResponsesClient()
|
||||
|
||||
# Create a hosted shell tool with the default auto container environment
|
||||
shell_tool = client.get_shell_tool()
|
||||
|
||||
agent = Agent(
|
||||
client=client,
|
||||
instructions="You are a helpful assistant that can execute shell commands to answer questions.",
|
||||
tools=shell_tool,
|
||||
)
|
||||
|
||||
query = "Use a shell command to show the current date and time"
|
||||
print(f"User: {query}")
|
||||
result = await agent.run(query)
|
||||
print(f"Result: {result}\n")
|
||||
|
||||
# Print shell-specific content details
|
||||
for message in result.messages:
|
||||
shell_calls = [c for c in message.contents if c.type == "shell_tool_call"]
|
||||
shell_results = [c for c in message.contents if c.type == "shell_tool_result"]
|
||||
|
||||
if shell_calls:
|
||||
print(f"Shell commands: {shell_calls[0].commands}")
|
||||
if shell_results and shell_results[0].outputs:
|
||||
for output in shell_results[0].outputs:
|
||||
if output.stdout:
|
||||
print(f"Stdout: {output.stdout}")
|
||||
if output.stderr:
|
||||
print(f"Stderr: {output.stderr}")
|
||||
if output.exit_code is not None:
|
||||
print(f"Exit code: {output.exit_code}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user