diff --git a/python/packages/core/agent_framework/_harness/_mode.py b/python/packages/core/agent_framework/_harness/_mode.py index 11cc018b178..9d8e1650286 100644 --- a/python/packages/core/agent_framework/_harness/_mode.py +++ b/python/packages/core/agent_framework/_harness/_mode.py @@ -8,18 +8,26 @@ from .._sessions import AgentSession, ContextProvider, SessionContext from .._telemetry import FeatureIndex, mark_feature_used -from .._tools import tool +from .._tools import FunctionTool, tool from .._types import Message DEFAULT_MODE_SOURCE_ID = "agent_mode" +_MODE_GET_INSTRUCTIONS = "Use the mode_get tool to check your current operating mode.\n" +_MODE_SET_INSTRUCTIONS = ( + "Use the mode_set tool to switch between modes as your work progresses. " + "Only use mode_set if the user explicitly instructs/allows you to change modes.\n\n" +) +_PLAN_MODE_TRANSITION = ( + "7. When approval is granted, always switch to execute mode (using the `mode_set` tool), " + "and follow the steps for *Execute mode*." +) DEFAULT_MODE_INSTRUCTIONS = ( "## Agent Mode\n\n" "- You can operate in different modes. Depending on the mode you are in, " "you will be required to follow different processes.\n\n" - "Use the mode_get tool to check your current operating mode.\n" - "Use the mode_set tool to switch between modes as your work progresses. " - "Only use mode_set if the user explicitly instructs/allows you to change modes.\n\n" - "You are currently operating in the {current_mode} mode.\n\n" + + _MODE_GET_INSTRUCTIONS + + _MODE_SET_INSTRUCTIONS + + "You are currently operating in the {current_mode} mode.\n\n" "### Mandatory Mode based Workflow\n\n" "For every new substantive user request, including short factual questions, " "your behavior is determined by the mode you are in.\n\n" @@ -49,8 +57,7 @@ "5. Write the plan to a memory file, so that it is retained even if compaction happens. " "Make sure to update the plan file if the user requests changes.\n" "6. Present the plan to the user and ask for approval to switch to execute mode and process the plan.\n" - "7. When approval is granted, always switch to execute mode (using the `mode_set` tool), " - "and follow the steps for *Execute mode*." + + _PLAN_MODE_TRANSITION ), "execute": ( "Determine the type of ask:\n" @@ -157,6 +164,7 @@ def set_agent_mode( *, source_id: str = DEFAULT_MODE_SOURCE_ID, available_modes: Sequence[str] | None = None, + notify: bool = True, ) -> str: """Set the current operating mode in session state. @@ -173,6 +181,9 @@ def set_agent_mode( Keyword Args: source_id: Unique source ID for the provider state. available_modes: Supported modes to validate against. Defaults to the built-in modes. + notify: Whether to notify the agent about the mode change on its next run. Set to ``False`` when the + agent changes mode through a replacement tool and has already observed the tool result. This also + clears any pending external-change notification. Returns: The normalized mode string that was stored. @@ -189,8 +200,11 @@ def set_agent_mode( # prior mode so the next ``before_run`` can inject a user message announcing the switch. Without # that injection, the model often anchors on the earlier ``set_mode`` tool call in the chat history # and keeps behaving as if it were still in that mode — system instructions alone are insufficient. - if isinstance(previous_mode, str) and previous_mode != normalized_mode: - provider_state[_PREVIOUS_MODE_STATE_KEY] = previous_mode + if notify: + if isinstance(previous_mode, str) and previous_mode != normalized_mode: + provider_state[_PREVIOUS_MODE_STATE_KEY] = previous_mode + else: + provider_state.pop(_PREVIOUS_MODE_STATE_KEY, None) return normalized_mode @@ -204,10 +218,13 @@ class AgentModeProvider(ContextProvider): The set of available modes is configurable with ``mode_instructions``. By default, two modes are provided: ``"plan"`` (interactive planning) and ``"execute"`` (autonomous execution). - This provider exposes the following tools to the agent: + By default, this provider exposes the following tools to the agent: - ``mode_set``: Switch the agent's operating mode. - ``mode_get``: Retrieve the agent's current operating mode. + Set ``expose_mode_set`` or ``expose_mode_get`` to ``False`` to omit that tool while retaining mode state + and workflow instructions. Replacement tools can be supplied through the agent's ``tools`` argument. + Public helper functions ``get_agent_mode`` and ``set_agent_mode`` allow external code to programmatically read and change the mode. """ @@ -219,6 +236,8 @@ def __init__( default_mode: str | None = None, mode_instructions: Mapping[str, str] | None = None, instructions: str | None = None, + expose_mode_set: bool = True, + expose_mode_get: bool = True, ) -> None: """Initialize a new agent mode provider. @@ -229,15 +248,30 @@ def __init__( default_mode: Initial mode used when no mode is stored yet. When omitted, the first entry of ``mode_instructions`` is used. mode_instructions: Mapping of supported modes to instructions on when and how to use each mode. + Custom text is not rewritten when tools are hidden. instructions: Custom instructions for using the mode tools. The instructions can contain an ``{available_modes}`` placeholder for the configured list of modes and a ``{current_mode}`` placeholder for the currently active mode. When omitted, the provider uses a default set of instructions. + Default guidance reflects tool exposure; custom text is not rewritten when tools are hidden. + expose_mode_set: Whether to contribute the built-in ``mode_set`` tool. Defaults to ``True``. + When ``False``, the application controls mode changes, optionally through a replacement tool. + expose_mode_get: Whether to contribute the built-in ``mode_get`` tool. Defaults to ``True``. + The current mode remains available in the default instructions and through ``get_agent_mode``. Raises: ValueError: No modes are configured, or the default mode is not configured. """ super().__init__(source_id) - mode_instructions = dict(DEFAULT_MODE_MAP if mode_instructions is None else mode_instructions) + if mode_instructions is None: + mode_instructions = dict(DEFAULT_MODE_MAP) + if not expose_mode_set: + mode_instructions["plan"] = mode_instructions["plan"].replace( + _PLAN_MODE_TRANSITION, + "7. When approval is granted, use the application's configured mode-change mechanism to " + "transition to execute mode. Follow the steps for *Execute mode* only after the mode has changed.", + ) + else: + mode_instructions = dict(mode_instructions) self._mode_display_names = _normalize_available_modes(tuple(mode_instructions)) if not self._mode_display_names: raise ValueError("mode_instructions must contain at least one mode.") @@ -247,6 +281,8 @@ def __init__( self.available_modes = tuple(self._mode_display_names) self.default_mode = _resolve_default_mode(default_mode, available_modes=self._mode_display_names) self.instructions = instructions + self.expose_mode_set = expose_mode_set + self.expose_mode_get = expose_mode_get def _build_instructions(self, current_mode: str) -> str: """Build the mode guidance injected for the current session.""" @@ -255,6 +291,15 @@ def _build_instructions(self, current_mode: str) -> str: for mode, mode_instruction in self.mode_instructions.items() ) instructions = self.instructions or DEFAULT_MODE_INSTRUCTIONS + if not self.instructions: + if not self.expose_mode_get: + instructions = instructions.replace(_MODE_GET_INSTRUCTIONS, "") + if not self.expose_mode_set: + instructions = instructions.replace( + _MODE_SET_INSTRUCTIONS, + "Mode changes are controlled by the application. Use its configured mode-change mechanism " + "only when the user explicitly instructs/allows a mode change.\n\n", + ) return instructions.replace("{available_modes}", mode_lines).replace("{current_mode}", current_mode) async def before_run( @@ -289,11 +334,13 @@ async def before_run( @tool(name="mode_set", approval_mode="never_require") def mode_set(mode: str) -> str: """Switch the agent's operating mode.""" - # The agent invoked the tool itself, so it knows the mode just changed — bypass - # ``set_agent_mode`` to avoid triggering a notification message on the next turn. - normalized_mode = _normalize_mode(mode, available_modes=self._mode_display_names) - tool_state = _get_mode_state(session, source_id=self.source_id) - tool_state["current_mode"] = normalized_mode + normalized_mode = set_agent_mode( + session, + mode, + source_id=self.source_id, + available_modes=self.available_modes, + notify=False, + ) return json.dumps({"mode": normalized_mode, "message": f"Mode changed to '{normalized_mode}'."}) @tool(name="mode_get", approval_mode="never_require") @@ -311,7 +358,12 @@ def mode_get() -> str: self.source_id, [self._build_instructions(current_mode)], ) - context.extend_tools(self.source_id, [mode_set, mode_get]) + tools: list[FunctionTool] = [] + if self.expose_mode_set: + tools.append(mode_set) + if self.expose_mode_get: + tools.append(mode_get) + context.extend_tools(self.source_id, tools) if isinstance(previous_mode, str) and previous_mode != current_mode: # Inject a user-role message announcing the external mode change. System instructions # always render first in the chat history, so the agent can otherwise stay anchored to diff --git a/python/packages/core/tests/core/test_harness_agent.py b/python/packages/core/tests/core/test_harness_agent.py index 6535a6fc68f..41030e3bd08 100644 --- a/python/packages/core/tests/core/test_harness_agent.py +++ b/python/packages/core/tests/core/test_harness_agent.py @@ -24,6 +24,7 @@ FileAccessProvider, FileMemoryProvider, FileSystemAgentFileStore, + FunctionTool, InMemoryAgentFileStore, InMemoryHistoryProvider, Message, @@ -33,6 +34,9 @@ SkillsProvider, TodoProvider, create_harness_agent, + get_agent_mode, + set_agent_mode, + tool, ) from agent_framework._harness._agent import DEFAULT_HARNESS_INSTRUCTIONS, _assemble_instructions from agent_framework._harness._mode import AgentModeProvider @@ -133,6 +137,56 @@ def test_create_harness_agent_disable_mode() -> None: assert AgentModeProvider not in provider_types +async def test_create_harness_agent_with_replacement_mode_tool() -> None: + """Applications can replace a built-in mode tool without replacing provider state.""" + session = AgentSession(session_id="session-1") + mode_provider = AgentModeProvider(source_id="ui_mode", expose_mode_set=False) + + @tool + def update_mode(mode: str) -> str: + """Update the application's mode.""" + return set_agent_mode( + session, + mode, + source_id=mode_provider.source_id, + available_modes=mode_provider.available_modes, + notify=False, + ) + + agent = create_harness_agent( + client=_FakeChatClient(), + max_context_window_tokens=128_000, + max_output_tokens=16_384, + mode_provider=mode_provider, + tools=[update_mode], + agent_instructions="Use update_mode for approved mode changes.", + disable_todo=True, + disable_file_memory=True, + disable_web_search=True, + ) + _, options = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Start planning"])], + ) + tools = options["tools"] + assert isinstance(tools, list) + assert [mode_tool.name for mode_tool in tools if isinstance(mode_tool, FunctionTool)] == ["update_mode", "mode_get"] + assert "Use update_mode for approved mode changes." in options["instructions"] + assert "mode_set" not in options["instructions"] + replacement = next( + mode_tool for mode_tool in tools if isinstance(mode_tool, FunctionTool) and mode_tool.name == "update_mode" + ) + await replacement.invoke(arguments={"mode": "execute"}) + assert get_agent_mode(session, source_id=mode_provider.source_id) == "execute" + + updated_context, updated_options = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Continue"])], + ) + assert "You are currently operating in the execute mode." in updated_options["instructions"] + assert updated_context.context_messages.get(mode_provider.source_id, []) == [] + + def test_create_harness_agent_disable_file_memory() -> None: """disable_file_memory=True should exclude the FileMemoryProvider.""" agent = create_harness_agent( diff --git a/python/packages/core/tests/core/test_harness_mode.py b/python/packages/core/tests/core/test_harness_mode.py index c4c221f33fe..57120f1ffba 100644 --- a/python/packages/core/tests/core/test_harness_mode.py +++ b/python/packages/core/tests/core/test_harness_mode.py @@ -11,11 +11,13 @@ Agent, AgentModeProvider, AgentSession, + FunctionTool, Message, SupportsChatGetResponse, get_agent_mode, set_agent_mode, ) +from agent_framework._harness._mode import DEFAULT_MODE_INSTRUCTIONS, DEFAULT_MODE_MAP def _tool_by_name(tools: list[object], name: str) -> object: @@ -210,6 +212,132 @@ async def test_agent_mode_context_provider_updates_agent_mode( assert set_agent_mode(session, "plan", source_id=provider.source_id) == "plan" +@pytest.mark.parametrize("expose_mode_set", [True, False]) +@pytest.mark.parametrize("expose_mode_get", [True, False]) +async def test_agent_mode_provider_tool_exposure( + chat_client_base: SupportsChatGetResponse, expose_mode_set: bool, expose_mode_get: bool +) -> None: + """Tool exposure must match built-in guidance without disabling the mode workflow.""" + session = AgentSession(session_id="session-1") + provider = AgentModeProvider(expose_mode_set=expose_mode_set, expose_mode_get=expose_mode_get) + agent = Agent(client=chat_client_base, context_providers=[provider]) + + _, options = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Start planning"])], + ) + tools: list[object] = options.get("tools") or [] + expected_names = [ + name for name, exposed in (("mode_set", expose_mode_set), ("mode_get", expose_mode_get)) if exposed + ] + assert [tool.name for tool in tools if isinstance(tool, FunctionTool)] == expected_names + instructions = options["instructions"] + assert isinstance(instructions, str) + assert ("mode_set" in instructions) == expose_mode_set + assert ("mode_get" in instructions) == expose_mode_get + assert "### Mandatory Mode based Workflow" in instructions + assert "get user approval before proceeding" in instructions + assert "You are currently operating in the plan mode." in instructions + assert get_agent_mode(session) == "plan" + if not expose_mode_set: + assert "only after the mode has changed" in instructions + for mode_tool in tools: + assert isinstance(mode_tool, FunctionTool) + assert mode_tool.approval_mode == "never_require" + if mode_tool.name == "mode_set": + result = await mode_tool.invoke(arguments={"mode": "execute"}) + assert result[0].text is not None + assert json.loads(result[0].text) == {"mode": "execute", "message": "Mode changed to 'execute'."} + else: + result = await mode_tool.invoke() + assert result[0].text is not None + assert json.loads(result[0].text) == {"mode": "execute" if expose_mode_set else "plan"} + + +@pytest.mark.parametrize("instructions", [None, ""]) +def test_agent_mode_provider_preserves_default_instructions(instructions: str | None) -> None: + """Suppression on one provider must not alter another provider's defaults.""" + AgentModeProvider(expose_mode_set=False, expose_mode_get=False) + provider = AgentModeProvider(instructions=instructions) + mode_lines = "".join(f"#### {name}\n\n{text}\n\n" for name, text in DEFAULT_MODE_MAP.items()) + expected = DEFAULT_MODE_INSTRUCTIONS.replace("{available_modes}", mode_lines).replace("{current_mode}", "plan") + assert provider._build_instructions("plan") == expected # pyright: ignore[reportPrivateUsage] + + +@pytest.mark.parametrize("custom_instructions", [True, False]) +@pytest.mark.parametrize("custom_modes", [True, False]) +def test_agent_mode_provider_preserves_custom_instructions(custom_instructions: bool, custom_modes: bool) -> None: + """Only built-in guidance should be adapted; caller text still expands placeholders.""" + mode_text = "Custom mode_get and mode_set guidance for {current_mode}." + mode_map = {"draft": mode_text} if custom_modes else None + instructions = "Use update_mode, not mode_set or mode_get. {current_mode}\n{available_modes}" + provider = AgentModeProvider( + expose_mode_set=False, + expose_mode_get=False, + instructions=instructions if custom_instructions else None, + mode_instructions=mode_map, + ) + current_mode = "draft" if custom_modes else "plan" + rendered = provider._build_instructions(current_mode) # pyright: ignore[reportPrivateUsage] + if custom_modes: + assert mode_map == {"draft": mode_text} + assert f"Custom mode_get and mode_set guidance for {current_mode}." in rendered + if custom_instructions: + assert rendered.startswith(f"Use update_mode, not mode_set or mode_get. {current_mode}\n") + assert "{available_modes}" not in rendered + assert "{current_mode}" not in rendered + if not custom_modes and not custom_instructions: + assert "mode_set" not in rendered + assert "mode_get" not in rendered + + +async def test_agent_mode_provider_hidden_tools_preserve_state_and_notifications( + chat_client_base: SupportsChatGetResponse, +) -> None: + """State and one-shot external notifications must survive even with no mode tools.""" + session = AgentSession(session_id="session-1") + provider = AgentModeProvider( + source_id="ui_mode", default_mode="execute", expose_mode_set=False, expose_mode_get=False + ) + agent = Agent(client=chat_client_base, context_providers=[provider]) + _, first_options = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Start"])], + ) + assert not first_options.get("tools") + assert "You are currently operating in the execute mode." in first_options["instructions"] + assert get_agent_mode(session, source_id=provider.source_id) == "execute" + set_agent_mode(session, "plan", source_id=provider.source_id) + + changed_context, changed_options = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Continue"])], + ) + assert "You are currently operating in the plan mode." in changed_options["instructions"] + notifications = changed_context.context_messages.get(provider.source_id, []) + assert len(notifications) == 1 + assert notifications[0].role == "user" + assert 'from "execute" to "plan"' in notifications[0].text + assert "previous_mode_for_notification" not in session.state[provider.source_id] + + set_agent_mode(session, "plan", source_id=provider.source_id) + unchanged_context, _ = await agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Continue planning"])], + ) + assert unchanged_context.context_messages.get(provider.source_id, []) == [] + + reconfigured_provider = AgentModeProvider(source_id=provider.source_id, default_mode="execute") + reconfigured_agent = Agent(client=chat_client_base, context_providers=[reconfigured_provider]) + _, restored_options = await reconfigured_agent._prepare_session_and_messages( # pyright: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Status"])], + ) + assert "You are currently operating in the plan mode." in restored_options["instructions"] + assert get_agent_mode(session, source_id=provider.source_id) == "plan" + assert DEFAULT_MODE_SOURCE_ID not in session.state + + def test_default_mode_falls_back_to_first_available_mode() -> None: """When ``default_mode`` is omitted, helpers and provider should use the first configured mode.""" session = AgentSession(session_id="session-1") @@ -251,6 +379,29 @@ def test_set_agent_mode_no_op_does_not_record_previous_mode() -> None: assert "previous_mode_for_notification" not in session.state[DEFAULT_MODE_SOURCE_ID] +def test_set_agent_mode_can_skip_external_change_notification() -> None: + """Agent-invoked replacement tools should be able to avoid a redundant notification.""" + session = AgentSession(session_id="session-1") + set_agent_mode(session, "plan") + set_agent_mode(session, "execute", notify=False) + + assert get_agent_mode(session) == "execute" + assert "previous_mode_for_notification" not in session.state[DEFAULT_MODE_SOURCE_ID] + + +def test_set_agent_mode_without_notification_clears_pending_notification() -> None: + """An agent-observed update should replace pending external transition context.""" + session = AgentSession(session_id="session-1") + set_agent_mode(session, "plan") + set_agent_mode(session, "execute") + assert session.state[DEFAULT_MODE_SOURCE_ID]["previous_mode_for_notification"] == "plan" + + set_agent_mode(session, "plan", notify=False) + + assert get_agent_mode(session) == "plan" + assert "previous_mode_for_notification" not in session.state[DEFAULT_MODE_SOURCE_ID] + + async def test_agent_mode_provider_injects_user_message_after_external_change( chat_client_base: SupportsChatGetResponse, ) -> None: diff --git a/python/samples/02-agents/context_providers/README.md b/python/samples/02-agents/context_providers/README.md index df0a3a3b1cc..1c8b5990146 100644 --- a/python/samples/02-agents/context_providers/README.md +++ b/python/samples/02-agents/context_providers/README.md @@ -57,3 +57,30 @@ These samples demonstrate how to use context providers to enrich agent conversat - Azure CLI authentication (`az login`) See each subfolder's README for provider-specific prerequisites. + +## Application-controlled modes + +Configure `AgentModeProvider(expose_mode_set=False)` to hide the built-in setter +when your application owns mode changes. Use `expose_mode_get=False` to hide the +getter independently, or set both flags to `False` to expose neither tool. Both +flags default to `True`. Mode state, per-turn workflow instructions, and external +mode-change notifications remain active even when both tools are hidden. + +Pass the configured provider through `create_harness_agent(mode_provider=...)` +(or `Agent(context_providers=[...])`). Supply any replacement tool, such as +`update_mode`, through the existing `tools` argument. Keep that tool and your UI +on the same session-backed state by using `get_agent_mode` and `set_agent_mode` +with the provider's `source_id` and `available_modes`, and its `default_mode` when +reading. A replacement tool should call `set_agent_mode(..., notify=False)` +because the agent already observes the tool result; this also clears any pending +external-change notification. UI-driven changes retain the default `notify=True` +so the agent sees the external change on its next run. Do not disable the entire +provider with `disable_mode=True`. + +Built-in guidance only advertises enabled tools. Without the built-in setter, +it defers approved transitions to the application's configured mode-change +mechanism. Supply replacement-specific guidance through the existing +`instructions` and `mode_instructions` options; caller-supplied text is not +rewritten when a tool is hidden, and existing placeholders still expand. +These flags only omit tools contributed by this provider: they do not filter +application-supplied tools or prevent application code from changing mode state.