diff --git a/python/packages/tools/agent_framework_tools/shell/_tool.py b/python/packages/tools/agent_framework_tools/shell/_tool.py index d85beeebd03..7f95dee259e 100644 --- a/python/packages/tools/agent_framework_tools/shell/_tool.py +++ b/python/packages/tools/agent_framework_tools/shell/_tool.py @@ -28,6 +28,7 @@ ) logger = logging.getLogger(__name__) +_audit_logger = logging.getLogger("agent_framework.shell.audit") def _quote_posix(value: str) -> str: @@ -143,6 +144,15 @@ class LocalShellTool: will execute any command the model emits. on_command: Optional audit hook called with the command string for every command that passes policy. Use for logging / telemetry. + + Note: + Shell execution events are emitted through + ``logging.getLogger("agent_framework.shell.audit")`` with ``shell.policy`` + (policy decisions) and ``shell.execution`` (outcomes) messages. + Configure a handler on that logger to collect, store, or forward events. + Commands are not included in log records by default because they may + contain secrets; use the ``on_command`` hook if you need them in a + controlled environment. """ def __init__( @@ -236,6 +246,9 @@ async def __aexit__(self, *_exc: object) -> None: # ------------------------------------------------------------------ core run + def _effective_workdir(self) -> str: + return os.path.abspath(self._workdir) if self._workdir is not None else os.getcwd() + async def run(self, command: str, *, timeout: float | None = None) -> ShellResult: """Execute ``command`` directly and return its :class:`ShellResult`. @@ -256,6 +269,19 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul mark_feature_used(FeatureIndex.TOOLS_SHELL) request = ShellRequest(command=command, workdir=self._workdir) decision = self._policy.evaluate(request) + # Resolve the effective cwd so records never report ``None``. + # In persistent mode the shell's real cwd may differ if + # ``confine_workdir=False``; this reports the configured location. + effective_workdir = self._effective_workdir() + _audit_logger.info( + "shell.policy", + extra={ + "policy_decision": decision.decision, + "denial_reason": decision.reason if decision.decision == "deny" else "", + "workdir": effective_workdir, + # command is intentionally omitted -- it may contain secrets. + }, + ) if decision.decision == "deny": raise ShellCommandError(f"Command rejected by policy: {decision.reason}") if self._on_command is not None: @@ -266,22 +292,43 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul effective_timeout = self._timeout if timeout is None else timeout - if self._mode == "persistent": - if self._session is None: - await self.start() - if self._session is None: - raise RuntimeError("LocalShellTool session failed to start") - effective = self._maybe_reanchor(command) - return await self._session.run(effective, timeout=effective_timeout) - - return await run_stateless( - self._stateless_argv, - command, - workdir=self._workdir, - env=self._env, - timeout=effective_timeout, - max_output_bytes=self._max_output_bytes, - ) + exit_code: int | None = None + timed_out = False + outcome = "unknown" + try: + if self._mode == "persistent": + if self._session is None: + await self.start() + if self._session is None: + raise RuntimeError("LocalShellTool session failed to start") + effective = self._maybe_reanchor(command) + result = await self._session.run(effective, timeout=effective_timeout) + else: + result = await run_stateless( + self._stateless_argv, + command, + workdir=self._workdir, + env=self._env, + timeout=effective_timeout, + max_output_bytes=self._max_output_bytes, + ) + exit_code = result.exit_code + timed_out = bool(getattr(result, "timed_out", False)) + outcome = "timeout" if timed_out else ("success" if exit_code == 0 else "failure") + return result + except BaseException as exc: + outcome = "cancelled" if isinstance(exc, asyncio.CancelledError) else "exception" + raise + finally: + _audit_logger.info( + "shell.execution", + extra={ + "exit_code": exit_code, + "timed_out": timed_out, + "outcome": outcome, + "workdir": effective_workdir, + }, + ) # ------------------------------------------------------------------ AF wiring diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py index f9dcbdf172c..0e89dd75996 100644 --- a/python/packages/tools/tests/test_local_shell_tool.py +++ b/python/packages/tools/tests/test_local_shell_tool.py @@ -2,6 +2,7 @@ import asyncio import json +import logging import os import shlex import shutil @@ -846,3 +847,68 @@ async def test_persistent_confines_workdir_by_default_powershell(tmp_path: os.Pa await tool.run(f"Set-Location -LiteralPath '{subdir}'") pwd = await tool.run("(Get-Location).Path") assert os.path.realpath(pwd.stdout.strip()) == os.path.realpath(str(tmp_path)) + + +def _audit_records(caplog: pytest.LogCaptureFixture, message: str) -> list[Any]: + return [r for r in caplog.records if r.name == "agent_framework.shell.audit" and r.message == message] + + +async def test_shell_logging_records_policy_decision_for_allowed_command( + caplog: pytest.LogCaptureFixture, +) -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi" + with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"): + await tool.run(cmd) + records = _audit_records(caplog, "shell.policy") + assert len(records) == 1 + assert records[0].__dict__["policy_decision"] == "allow" + assert records[0].__dict__["denial_reason"] == "" + assert "command" not in records[0].__dict__ + + +async def test_shell_logging_records_execution_outcome(caplog: pytest.LogCaptureFixture) -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi" + with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"): + await tool.run(cmd) + records = _audit_records(caplog, "shell.execution") + assert len(records) == 1 + assert records[0].__dict__["exit_code"] == 0 + assert records[0].__dict__["timed_out"] is False + assert records[0].__dict__["outcome"] == "success" + + +async def test_shell_logging_records_denial(caplog: pytest.LogCaptureFixture) -> None: + tool = LocalShellTool( + mode="stateless", + approval_mode="never_require", + acknowledge_unsafe=True, + policy=ShellPolicy(denylist=[r".*"]), + ) + with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"), pytest.raises(ShellCommandError): + await tool.run("echo hi") + policy_records = _audit_records(caplog, "shell.policy") + assert len(policy_records) == 1 + assert policy_records[0].__dict__["policy_decision"] == "deny" + assert policy_records[0].__dict__["denial_reason"] + assert _audit_records(caplog, "shell.execution") == [] + + +async def test_shell_logging_effective_workdir_resolves_none(caplog: pytest.LogCaptureFixture) -> None: + """With no workdir configured, the logged workdir is the process cwd, not None.""" + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi" + with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"): + await tool.run(cmd) + records = _audit_records(caplog, "shell.policy") + assert records[0].__dict__["workdir"] == os.getcwd() + + +async def test_shell_logging_records_configured_workdir(caplog: pytest.LogCaptureFixture, tmp_path: Any) -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True, workdir=tmp_path) + cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi" + with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"): + await tool.run(cmd) + assert _audit_records(caplog, "shell.policy")[0].__dict__["workdir"] == os.path.abspath(tmp_path) + assert _audit_records(caplog, "shell.execution")[0].__dict__["workdir"] == os.path.abspath(tmp_path) diff --git a/python/samples/02-agents/harness/README.md b/python/samples/02-agents/harness/README.md index 20ccc49780d..474f1945d90 100644 --- a/python/samples/02-agents/harness/README.md +++ b/python/samples/02-agents/harness/README.md @@ -30,6 +30,7 @@ Each feature can be disabled or customized via keyword arguments. |------|-------------| | `harness_research.py` | Interactive research assistant with web search, a plan/execute workflow, and an execute-mode loop that re-invokes the agent until every todo is complete | | `harness_data_processing.py` | Data-processing assistant over a folder of CSV files, demonstrating file-access tools and tool approval | +| `async_request_broker.py` | MAF approval pause-and-resume: agent pauses for shell-command approval, application delivers the request to a simulated client, and resumes the agent with the decision | | [`build_your_own_claw/`](./build_your_own_claw/README.md) | *Build your own claw* blog series — a personal finance assistant built step by step | ## Running diff --git a/python/samples/02-agents/harness/async_request_broker.py b/python/samples/02-agents/harness/async_request_broker.py new file mode 100644 index 00000000000..ff35906771d --- /dev/null +++ b/python/samples/02-agents/harness/async_request_broker.py @@ -0,0 +1,228 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import logging +import sys +from collections.abc import Awaitable, Mapping, Sequence +from typing import Any + +from agent_framework import ( + Agent, + AgentSession, + BaseChatClient, + ChatResponse, + Content, + FunctionInvocationLayer, + Message, +) +from agent_framework_tools.shell import LocalShellTool + +"""MAF approval pause-and-resume for shell commands. + +Agent Framework's approval contract is pause-and-resume, not "block and wait": + +1. A tool created with ``approval_mode="always_require"`` does NOT run. The + agent returns early and ``response.user_input_requests`` contains a + function-approval request. +2. Your application delivers that request to whoever decides (a UI, a chat + channel, a reviewer) over a transport that YOU own. +3. When a decision arrives, call ``agent.run(...)`` again with the SAME + ``AgentSession`` and an approval response built with + ``request.to_function_approval_response(approved)``. + +The framework deliberately ships no request broker: where pending requests +live, how they reach a client, and how long they wait are application concerns. + +This sample shows that split. ``_PendingApprovals`` below is a minimal +``asyncio.Future``-based mailbox that stands in for your transport. It is +process-lifetime only: if the process exits, pending requests are lost. It is +NOT a substitute for a durable queue or workflow engine; use one of those when +approvals must survive restarts or span multiple workers. + +No LLM or credentials are needed. A scripted chat client emits a shell tool +call, and the real ``LocalShellTool`` runs it only after approval. Shell +policy/outcome events are emitted on the ``agent_framework.shell.audit`` +logger; the sample enables it so you can see them. +""" + + +class _AuditPrinter(logging.Handler): + """Print shell audit events (policy decision and outcome) with their structured fields.""" + + _FIELDS = ("policy_decision", "outcome", "exit_code", "timed_out") + + def emit(self, record: logging.LogRecord) -> None: + fields = {k: record.__dict__[k] for k in self._FIELDS if k in record.__dict__} + print(f" [audit] {record.getMessage()} {fields}") + + +_audit_logger = logging.getLogger("agent_framework.shell.audit") +_audit_logger.setLevel(logging.INFO) +_audit_logger.addHandler(_AuditPrinter()) + + +class _PendingApprovals: + """Application-owned mailbox: request id -> Future[bool]. Process-lifetime only.""" + + def __init__(self) -> None: + self._futures: dict[str, asyncio.Future[bool]] = {} + + def submit(self, request_id: str) -> asyncio.Future[bool]: + future: asyncio.Future[bool] = asyncio.get_running_loop().create_future() + self._futures[request_id] = future + return future + + def answer(self, request_id: str, approved: bool) -> None: + future = self._futures.pop(request_id, None) + if future is not None and not future.done(): + future.set_result(approved) + + def discard(self, request_id: str) -> None: + self._futures.pop(request_id, None) + + @property + def pending_count(self) -> int: + return len(self._futures) + + +class _ScriptedShellClient(FunctionInvocationLayer, BaseChatClient): + """Stands in for an LLM: asks to run one shell command, then summarizes.""" + + def __init__(self, command: str) -> None: + super().__init__() + self._command = command + self._asked = False + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse]: + async def _get() -> ChatResponse: + if not self._asked: + self._asked = True + call = Content.from_function_call( + call_id="call-1", name="run_shell", arguments={"command": self._command} + ) + return ChatResponse(messages=Message(role="assistant", contents=[call])) + return ChatResponse(messages=Message(role="assistant", contents=["The command finished."])) + + return _get() + + +def _make_agent(command: str) -> Agent[Any]: + shell = LocalShellTool(mode="stateless", approval_mode="always_require") + return Agent(client=_ScriptedShellClient(command), tools=[shell.as_function()]) + + +async def simulated_client(pending: _PendingApprovals, request_id: str, *, approve: bool, delay: float) -> None: + """Simulate a remote reviewer answering after a delay (or never, if delay is None-like).""" + await asyncio.sleep(delay) + print(f"[client] {'Approving' if approve else 'Denying'} {request_id}") + pending.answer(request_id, approve) + + +async def run_with_approval( + agent: Agent[Any], + session: AgentSession, + pending: _PendingApprovals, + prompt: str, + *, + answer_timeout: float, + client_task_factory: Any, +) -> str: + """Drive pause-and-resume. ``client_task_factory(request_id)`` starts the simulated client.""" + response = await agent.run(prompt, session=session) + if not response.user_input_requests: + return response.text + + request = response.user_input_requests[0] + assert request.function_call is not None + request_id = request.id or request.function_call.call_id + print(f"[server] Paused: approval needed for {request.function_call.name}({request.function_call.arguments})") + + # Deliver the request to the client over *your* transport, then wait for a decision. + decision = pending.submit(request_id) + client_task = client_task_factory(request_id) + try: + approved = await asyncio.wait_for(decision, timeout=answer_timeout) + except asyncio.TimeoutError: + # Nobody answered. The agent is NOT blocked -- it already returned -- but this + # conversation is stuck in a "pending approval" state until you act. Decide a + # policy: keep waiting, escalate, or resume with a denial. + print(f"[server] No answer within {answer_timeout}s; request {request_id} is still unresolved.") + pending.discard(request_id) + client_task.cancel() + return "STUCK: approval was never answered (session left paused)" + + # Resume: same session, plus the approval response. + print(f"[server] Resuming with approved={approved}") + resumed = await agent.run( + [ + Message("assistant", [request]), + Message("user", [request.to_function_approval_response(approved)]), + ], + session=session, + ) + return resumed.text + + +async def main() -> None: + command = "Write-Output hello" if sys.platform == "win32" else "echo hello from approved shell" + pending = _PendingApprovals() + + print("=== 1. Approved: the command runs after the client says yes ===") + agent, session = _make_agent(command), AgentSession() + print( + "[agent]", + await run_with_approval( + agent, + session, + pending, + "Run the greeting command", + answer_timeout=2.0, + client_task_factory=lambda rid: asyncio.create_task( + simulated_client(pending, rid, approve=True, delay=0.1) + ), + ), + ) + + print("\n=== 2. Denied: the command never runs ===") + agent, session = _make_agent(command), AgentSession() + print( + "[agent]", + await run_with_approval( + agent, + session, + pending, + "Run the greeting command", + answer_timeout=2.0, + client_task_factory=lambda rid: asyncio.create_task( + simulated_client(pending, rid, approve=False, delay=0.1) + ), + ), + ) + + print("\n=== 3. Nobody answers: the session is stuck pending ===") + agent, session = _make_agent(command), AgentSession() + print( + "[agent]", + await run_with_approval( + agent, + session, + pending, + "Run the greeting command", + answer_timeout=0.3, + client_task_factory=lambda rid: asyncio.create_task( + simulated_client(pending, rid, approve=True, delay=30.0) + ), + ), + ) + print(f"Pending requests: {pending.pending_count}") + + +if __name__ == "__main__": + asyncio.run(main())