agent-framework-tools 測試版 Python 套件透過命名空間提供 shell 執行與環境感知工具agent_framework.tools。
| Tool | 在下列情況下使用 |
|---|---|
LocalShellTool |
指令是受信任的或個別核准的,且應在代理程序的宿主環境中執行。 |
DockerShellTool |
模型生成的殼層指令需要 OCI 容器隔離。 |
ShellEnvironmentProvider |
該模型需要主動殼系列、作業系統、工作目錄,以及已安裝的 CLI 版本。 |
ShellPolicy |
你需要在批准或執行前先做允許清單或拒絕清單的預篩選。 |
Warning
殼層執行可以修改檔案、啟動程序、存取憑證,並與外部系統通訊。 使用支援該任務的最低權限執行層級。
安裝套件
dotnet add package Microsoft.Agents.AI.Tools.Shell --prerelease
利用當地殼層與環境感知
LocalShellExecutor 支援無狀態與持久模式。
ShellEnvironmentProvider 探測主動環境,並為代理上下文加入權威的殼層指引。
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 直接在主機上執行指令。 預設為持久殼層、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
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 探測殼殼族、版本、作業系統、工作目錄及選定的 CLI 版本,然後在代理程式執行前注入該資訊。 預設的探測清單為 git、 node、 python、 docker和 。
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(Docker)或 Podman 上。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 跑取樣文。
選擇執行層級
| 情境 | Tool | 隔離界限 |
|---|---|---|
| 受信任的開發指令 | LocalShellTool |
主機程序中的核准 |
| 不受信任的殼層指令 | DockerShellTool |
帶有預設隔離標誌的 OCI 容器 |
| 無 shell 的不可信產生程式碼 | 超光代碼法案 | Hyperlight microVM |
Go 提供本地 shell 執行與環境探測。tool/shelltool 請參見 「使用本地殼層工具」。
DockerShellTool 目前沒有針對 Go 的指引。
使用殼型工具搭配 Harness Agent
純代理,並 HarnessAgent 使用相同的兩部分 shell 設定:註冊執行者的功能作為工具,並在模型應該接收 shell、作業系統、工作目錄及 CLI 版本上下文時新增 ShellEnvironmentProvider 。
HarnessAgent 不建立或擁有殼執行者:
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、nodepythondocker、 ,每個探測有五秒的超時。
每個使用者會話建立一個持久執行器,並在會話結束時將其處置。 不要在使用者間或同時對話中分享,因為工作目錄、環境、shell 歷史、背景工作和指令佇列都是共享的。
ShellPolicy 僅為預過濾器;保持啟用批准,使用權限最低的憑證,且當指令需要更強的隔離邊界時,優先 DockerShellExecutor 使用。
Shell 工具可從預發布 Microsoft.Agents.AI.Tools.Shell 套件中取得。
HarnessAgent 可從 Microsoft.Agents.AI.Harness取得。
對於純代理,分別建立殼函數 和 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 工具核准,請將 傳給 AgentSessionrun。 呼叫者擁有執行者生命週期;使用 async with 或呼叫 close(),並在每個使用者會話中建立一個持久化工具。 不要在使用者間或同時對話中共享可變的 shell 狀態。
主機殼不是沙盒。 保持啟用批准,使用權限最低的憑證,並用於 DockerShellTool 容器隔離。 停用核准需要approval_mode="never_require"且acknowledge_unsafe=TrueShellPolicy;僅此本身不是安全界線。
create_harness_agent 在 中釋出 agent-framework-core。 Shell 整合由預發布agent-framework-tools套件提供,啟用時會發出 。ExperimentalWarning
目前沒有包裝的 Go 安全帶。 直接在 Classic Go 代理上撰寫本地 shell 工具和環境服務。