beta agent-framework-tools Python 包通过agent_framework.tools命名空间提供 shell 执行和环境感知工具。
| 工具 | 在以下情况下使用 |
|---|---|
LocalShellTool |
命令受信任或单独批准,应在代理进程的主机环境中运行。 |
DockerShellTool |
模型生成的 shell 命令需要 OCI 容器隔离。 |
ShellEnvironmentProvider |
该模型需要活动的 shell 系列、操作系统、工作目录和已安装的 CLI 版本。 |
ShellPolicy |
在审批或执行之前,需要允许列表或拒绝列表预筛选。 |
Warning
Shell 执行可以修改文件、启动进程、访问凭据并与外部系统通信。 使用支持该任务的最小特权执行层。
安装软件包
dotnet add package Microsoft.Agents.AI.Tools.Shell --prerelease
使用本地 shell 和环境感知
LocalShellExecutor 支持无状态模式和持久性模式。
ShellEnvironmentProvider 探测活动环境,并向代理上下文添加权威 shell 指南。
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Tools.Shell;
using Microsoft.Extensions.AI;
var endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT") ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-5.4-mini";
// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid
// latency issues, unintended credential probing, and potential security risks from fallback mechanisms.
var aiProjectClient = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential());
const string Instructions = """
You are an agent with a single tool: run_shell. Use it to satisfy the
user's request. Do not describe what you would do — actually run the
commands. Reply with the final answer derived from real output.
""";
// --------------------------------------------------------------------
// 1. Stateless mode — each call gets a fresh shell.
// --------------------------------------------------------------------
Console.WriteLine("### Stateless mode\n");
await using (var statelessShell = new LocalShellExecutor(new() { Mode = ShellMode.Stateless, AcknowledgeUnsafe = true }))
{
var envProvider = new ShellEnvironmentProvider(statelessShell);
var statelessAgent = aiProjectClient.AsAIAgent(new ChatClientAgentOptions
{
ChatOptions = new()
{
ModelId = deploymentName,
Instructions = Instructions,
Tools = [statelessShell.AsAIFunction(requireApproval: false)],
},
AIContextProviders = [envProvider],
});
// --------------------------------------------------------------------
// 2. Persistent mode — one shell, reused across calls. State carries.
// --------------------------------------------------------------------
Console.WriteLine("\n### Persistent mode\n");
await using (var persistentShell = new LocalShellExecutor(new() { Mode = ShellMode.Persistent, AcknowledgeUnsafe = true }))
{
var envProvider = new ShellEnvironmentProvider(persistentShell);
var persistentAgent = aiProjectClient.AsAIAgent(new ChatClientAgentOptions
{
ChatOptions = new()
{
ModelId = deploymentName,
Instructions = Instructions,
Tools = [persistentShell.AsAIFunction(requireApproval: false)],
},
AIContextProviders = [envProvider],
});
var persistentSession = await persistentAgent.CreateSessionAsync();
// State carries across calls in persistent mode: cd into temp, then
// verify the next call sees the new CWD.
Console.WriteLine(await persistentAgent.RunAsync("Change directory into the system temp folder, then print the current working directory.", persistentSession));
Console.WriteLine();
Console.WriteLine(await persistentAgent.RunAsync("In a NEW shell call, print the current working directory again. Tell me whether it still matches the temp folder.", persistentSession));
Console.WriteLine();
// Same idea with an exported variable: set in one call, read in the next.
Console.WriteLine(await persistentAgent.RunAsync("Set the environment variable DEMO_TOKEN to the value 'hello-world'.", persistentSession));
Console.WriteLine();
Console.WriteLine(await persistentAgent.RunAsync("Print the current value of DEMO_TOKEN. Tell me exactly what value the shell reports.", persistentSession));
Console.WriteLine();
PrintSnapshot(envProvider.CurrentSnapshot!);
}
ShellPolicy 也可用于命令预筛选。 当前未发布专用的可 DockerShellExecutor 运行示例。
安装软件包
pip install agent-framework-tools --pre
包安装 psutil 以在执行超时时终止子进程树。
使用 LocalShellTool
LocalShellTool 直接在主机上运行命令。 它默认为持久性 shell、30 秒超时、64-KiB 输出截断、工作目录限制和每个命令的审批。
import asyncio
from typing import Any
from agent_framework import Agent, Message
from agent_framework.openai import OpenAIChatClient
from agent_framework.tools import LocalShellTool
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
async def main() -> None:
print("=== OpenAI Agent with LocalShellTool Example ===")
print("NOTE: Commands will execute on your local machine.\n")
client = OpenAIChatClient(model="gpt-5.4-nano")
async with LocalShellTool() as shell:
agent = Agent(
client=client,
instructions="You are a helpful assistant that can run shell commands to help the user.",
tools=[client.get_shell_tool(func=shell.as_function())],
)
query = "Use the shell 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:
if user_input_needed.function_call is None:
continue
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())
当每个调用应在新的进程中运行时使用 mode="stateless" 。
AGENT_FRAMEWORK_SHELL使用环境变量或shell构造函数参数替代解析的 shell。
Important
LocalShellTool 不是沙盒。 审批是主要安全边界。 禁用审批需要 acknowledge_unsafe=True。
使用 限制命令
ShellPolicy 在执行前应用正则表达式允许和拒绝列表。 拒绝规则优先。
import asyncio
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from agent_framework.tools import LocalShellTool, ShellPolicy
from dotenv import load_dotenv
load_dotenv()
async def main() -> None:
client = OpenAIChatClient(model="gpt-5.4-nano")
shell = LocalShellTool(
mode="stateless",
approval_mode="never_require",
acknowledge_unsafe=True,
policy=ShellPolicy(
allowlist=[
r"^ls(\s|$)",
r"^pwd$",
r"^cat\s[^|;&]+$",
r"^git\s+(status|log|diff)(\s|$)",
r"^python\s+--version$",
],
),
timeout=10,
)
agent = Agent(
client=client,
instructions=(
"You can run a narrow set of read-only shell commands (ls, pwd, cat, "
"git status/log/diff, python --version). Anything else will be rejected."
),
tools=[client.get_shell_tool(func=shell.as_function())],
)
query = "Summarise the current directory and print the Python version."
print(f"User: {query}")
result = await agent.run(query)
print(f"Agent: {result.text}")
Warning
命令策略是可用性预筛选,而不是安全边界。 Shell 语法、别名、变量、解释器和编码的有效负载可以绕过简单的模式匹配。
添加 ShellEnvironmentProvider
ShellEnvironmentProvider 探测 shell 系列、版本、操作系统、工作目录和所选 CLI 版本,然后在代理运行之前注入该信息。 默认探测列表为git、node和pythondocker。
import asyncio
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from agent_framework.tools import (
LocalShellTool,
ShellEnvironmentProvider,
ShellEnvironmentProviderOptions,
)
from dotenv import load_dotenv
load_dotenv()
def _print_snapshot(label: str, provider: ShellEnvironmentProvider) -> None:
snapshot = provider.current_snapshot
if snapshot is None:
print(f"[{label}] no snapshot captured")
return
print(f"\n[{label}] snapshot:")
print(f" family = {snapshot.family.value}")
print(f" os = {snapshot.os_description}")
print(f" shell_version = {snapshot.shell_version}")
print(f" working_directory = {snapshot.working_directory}")
for tool, version in snapshot.tool_versions.items():
print(f" {tool:<17} = {version}")
async def _ask(agent: Agent, query: str) -> None:
print(f"\nUser: {query}")
result = await agent.run(query)
if result.text:
print(f"Agent: {result.text}")
async def main() -> None:
client = OpenAIChatClient(model="gpt-5.4-nano")
options = ShellEnvironmentProviderOptions(
probe_tools=("git", "python", "uv", "node"),
)
print("=== stateless mode ===")
async with LocalShellTool(
mode="stateless",
approval_mode="never_require",
acknowledge_unsafe=True,
) as shell:
provider = ShellEnvironmentProvider(shell, options)
agent = Agent(
client=client,
instructions="Use the shell tool to answer the user's question.",
tools=[client.get_shell_tool(func=shell.as_function())],
context_providers=[provider],
)
await _ask(agent, "Show me the current working directory.")
await _ask(agent, "Now `cd ..` then show the working directory again.")
await _ask(agent, "Show the working directory once more — did `cd` persist?")
_print_snapshot("stateless", provider)
print("\n=== persistent mode ===")
async with LocalShellTool(
mode="persistent",
confine_workdir=False,
approval_mode="never_require",
acknowledge_unsafe=True,
) as shell:
provider = ShellEnvironmentProvider(shell, options)
agent = Agent(
client=client,
instructions="Use the shell tool to answer the user's question.",
tools=[client.get_shell_tool(func=shell.as_function())],
context_providers=[provider],
)
await _ask(agent, "Show me the current working directory.")
await _ask(agent, "Now `cd ..` then show the working directory again.")
await _ask(agent, "Show the working directory once more — did `cd` persist?")
_print_snapshot("persistent", provider)
使用 DockerShellTool
DockerShellTool 需要 Docker 或 Podman on PATH。 默认值禁用网络,以非根用户身份运行,使用只读根文件系统,删除功能,将内存限制为 512 MiB,并将容器上限为 256 个进程。
from agent_framework.tools import DockerShellTool
async with DockerShellTool(
image="mcr.microsoft.com/azurelinux/base/core:3.0",
approval_mode="never_require",
) as shell:
result = await shell.run("uname -a && id")
print(result.stdout)
默认图像为 mcr.microsoft.com/azurelinux/base/core:3.0. 传递给 docker_binary="podman" 使用 Podman。 当前未发布专用的可 DockerShellTool 运行示例。
选择执行层
| 情景 | 工具 | 隔离边界 |
|---|---|---|
| 受信任的开发命令 | LocalShellTool |
主机进程中的审批 |
| 不受信任的 shell 命令 | DockerShellTool |
具有默认隔离标志的 OCI 容器 |
| 没有 shell 的不受信任的生成的代码 | Hyperlight CodeAct | Hyperlight microVM |
Go 提供本地 shell 执行和环境探测。tool/shelltool 请参阅 使用本地 shell 工具。
DockerShellTool 指南当前不适用于 Go。
将 shell 工具与 Harness 代理配合使用
纯代理并使用 HarnessAgent 同一个由两部分构成的 shell 设置:将执行程序的函数注册为工具,并在模型应接收 shell、操作系统、工作目录和 CLI 版本上下文时添加 ShellEnvironmentProvider 。
HarnessAgent 不创建或拥有 shell 执行程序:
using System.IO;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Tools.Shell;
using Microsoft.Extensions.AI;
await using var shell = new LocalShellExecutor(new LocalShellExecutorOptions
{
WorkingDirectory = Directory.GetCurrentDirectory(),
Timeout = LocalShellExecutor.DefaultTimeout,
});
AIAgent agent = chatClient.AsHarnessAgent(new HarnessAgentOptions
{
AIContextProviders = [new ShellEnvironmentProvider(shell)],
ChatOptions = new ChatOptions
{
Tools = [shell.AsAIFunction(requireApproval: true)],
},
});
AsAIFunction 默认为名称和 run_shellrequireApproval: true。
LocalShellExecutor 默认为永久性模式,每个输出流 64-KiB 上限,且无超时;该示例显式使用建议的 30 秒 LocalShellExecutor.DefaultTimeout。
ShellEnvironmentProviderOptions 默认为探测 git、 dotnet、 node和 python,每个 docker探测的超时为 5 秒。
为每个用户会话创建一个永久性执行程序,并在会话结束时释放它。 不要在用户或并发对话之间共享,因为工作目录、环境、shell 历史记录、后台作业和命令队列是共享的。
ShellPolicy 只是预筛选;保持审批启用状态,使用最低特权凭据,并在命令需要更强大的隔离边界时优先 DockerShellExecutor 使用。
预发行版 Microsoft.Agents.AI.Tools.Shell 包中提供了 Shell 工具。
HarnessAgent 可从中获取 Microsoft.Agents.AI.Harness。
对于纯代理,请使用 创建 shell 函数 client.get_shell_tool(func=shell.as_function()) 并单独添加 ShellEnvironmentProvider 。
create_harness_agent 通过 shell_executor时执行这两个步骤:
from agent_framework import create_harness_agent
from agent_framework.tools import LocalShellTool, ShellEnvironmentProviderOptions
async with LocalShellTool() as shell:
agent = create_harness_agent(
client=client,
shell_executor=shell,
shell_environment_provider_options=ShellEnvironmentProviderOptions(
probe_tools=("git", "python"),
),
)
session = agent.create_session()
response = await agent.run("Inspect the current repository.", session=session)
shell_executor 是选择加入,必须公开 as_function()。 工厂将添加 shell 工具, ShellEnvironmentProvider 并且仅当客户端实现 SupportsShellTool时;否则会记录警告并跳过这两者。
shell_environment_provider_options 是可选的,仅用于 shell_executor.
LocalShellTool 默认为持久性模式、30 秒超时、64 KiB 组合输出、工作目录重新定位和 approval_mode="always_require"。 由于 Harness 工具审批默认处于启用状态,因此请传递一个 AgentSession 到 run。 调用方拥有执行程序生命周期;使用 async with 或调用 close(),并为每个用户会话创建一个持久性工具。 不要在用户或并发对话之间共享可变 shell 状态。
主机外壳不是沙盒。 保持审批启用状态,使用最低特权凭据,并用于 DockerShellTool 容器隔离。 禁用审批需要approval_mode="never_require";acknowledge_unsafe=TrueShellPolicy单独不是安全边界。
create_harness_agent 在 .. 中 agent-framework-core发布。 Shell 集成由预发行 agent-framework-tools 包提供,并在启用时发出 ExperimentalWarning 。
打包的 Go Harness 当前不可用。 直接在普通 Go 代理上编写本地 shell 工具和环境提供程序。