Python: Updated instructions/system_message logic in GitHub Copilot agent

python/packages/github_copilot/agent_framework_github_copilot/_agent.py

@@ -31,6 +31,7 @@
     PermissionRequestResult,
     ResumeSessionConfig,
     SessionConfig,
+    SystemMessageConfig,
     ToolInvocation,
     ToolResult,
 )
@@ -57,8 +58,9 @@
 class GitHubCopilotOptions(TypedDict, total=False):
     """GitHub Copilot-specific options."""
 
-    instructions: str
-    """System message to append to the session."""
+    system_message: SystemMessageConfig
+    """System message configuration for the session. Use mode 'append' to add to the default
+    system prompt, or 'replace' to completely override it."""
 
     cli_path: str
     """Path to the Copilot CLI executable. Defaults to GITHUB_COPILOT_CLI_PATH environment variable
@@ -139,6 +141,7 @@ def get_weather(city: str) -> str:
 
     def __init__(
         self,
+        instructions: str | None = None,
         *,
         client: CopilotClient | None = None,
         id: str | None = None,
@@ -157,6 +160,9 @@ def __init__(
     ) -> None:
         """Initialize the GitHub Copilot Agent.
 
+        Args:
+            instructions: System message for the agent.
+
         Keyword Args:
             client: Optional pre-configured CopilotClient instance. If not provided,
                 a new client will be created using the other parameters.
@@ -188,7 +194,10 @@ def __init__(
 
         # Parse options
         opts: dict[str, Any] = dict(default_options) if default_options else {}
-        instructions = opts.pop("instructions", None)
+
+        # Handle instructions - direct parameter takes precedence over default_options.system_message
+        self._prepare_system_message(instructions, opts)
+
         cli_path = opts.pop("cli_path", None)
         model = opts.pop("model", None)
         timeout = opts.pop("timeout", None)
@@ -208,7 +217,6 @@ def __init__(
         except ValidationError as ex:
             raise ServiceInitializationError("Failed to create GitHub Copilot settings.", ex) from ex
 
-        self._instructions = instructions
         self._tools = normalize_tools(tools)
         self._permission_handler = on_permission_request
         self._mcp_servers = mcp_servers
@@ -302,7 +310,7 @@ async def run(
         opts: dict[str, Any] = dict(options) if options else {}
         timeout = opts.pop("timeout", None) or self._settings.timeout or DEFAULT_TIMEOUT_SECONDS
 
-        session = await self._get_or_create_session(thread, streaming=False)
+        session = await self._get_or_create_session(thread, streaming=False, runtime_options=opts)
         input_messages = normalize_messages(messages)
         prompt = "\n".join([message.text for message in input_messages])
 
@@ -365,7 +373,9 @@ async def run_stream(
         if not thread:
             thread = self.get_new_thread()
 
-        session = await self._get_or_create_session(thread, streaming=True)
+        opts: dict[str, Any] = dict(options) if options else {}
+
+        session = await self._get_or_create_session(thread, streaming=True, runtime_options=opts)
         input_messages = normalize_messages(messages)
         prompt = "\n".join([message.text for message in input_messages])
 
@@ -400,6 +410,29 @@ def event_handler(event: SessionEvent) -> None:
         finally:
             unsubscribe()
 
+    @staticmethod
+    def _prepare_system_message(
+        instructions: str | None,
+        opts: dict[str, Any],
+    ) -> None:
+        """Prepare system message configuration in opts.
+
+        If instructions is provided, it takes precedence for content.
+        If system_message is also provided, its mode is preserved.
+        Modifies opts in place.
+
+        Args:
+            instructions: Direct instructions parameter for content.
+            opts: Options dictionary to modify.
+        """
+        opts_system_message = opts.pop("system_message", None)
+        if instructions is not None:
+            # Use instructions for content, but preserve mode from system_message if provided
+            mode = opts_system_message.get("mode", "append") if opts_system_message else "append"
+            opts["system_message"] = {"mode": mode, "content": instructions}
+        elif opts_system_message is not None:
+            opts["system_message"] = opts_system_message
+
     def _prepare_tools(
         self,
         tools: list[ToolProtocol | MutableMapping[str, Any]],
@@ -459,12 +492,14 @@ async def _get_or_create_session(
         self,
         thread: AgentThread,
         streaming: bool = False,
+        runtime_options: dict[str, Any] | None = None,
     ) -> CopilotSession:
         """Get an existing session or create a new one for the thread.
 
         Args:
             thread: The conversation thread.
             streaming: Whether to enable streaming for the session.
+            runtime_options: Runtime options from run/run_stream that take precedence.
 
         Returns:
             A CopilotSession instance.
@@ -479,33 +514,47 @@ async def _get_or_create_session(
             if thread.service_thread_id:
                 return await self._resume_session(thread.service_thread_id, streaming)
 
-            session = await self._create_session(streaming)
+            session = await self._create_session(streaming, runtime_options)
             thread.service_thread_id = session.session_id
             return session
         except Exception as ex:
             raise ServiceException(f"Failed to create GitHub Copilot session: {ex}") from ex
 
-    async def _create_session(self, streaming: bool) -> CopilotSession:
-        """Create a new Copilot session."""
+    async def _create_session(
+        self,
+        streaming: bool,
+        runtime_options: dict[str, Any] | None = None,
+    ) -> CopilotSession:
+        """Create a new Copilot session.
+
+        Args:
+            streaming: Whether to enable streaming for the session.
+            runtime_options: Runtime options that take precedence over default_options.
+        """
         if not self._client:
             raise ServiceException("GitHub Copilot client not initialized. Call start() first.")
 
+        opts = runtime_options or {}
         config: SessionConfig = {"streaming": streaming}
 
-        if self._settings.model:
-            config["model"] = self._settings.model  # type: ignore[typeddict-item]
+        model = opts.get("model") or self._settings.model
+        if model:
+            config["model"] = model  # type: ignore[typeddict-item]
 
-        if self._instructions:
-            config["system_message"] = {"mode": "append", "content": self._instructions}
+        system_message = opts.get("system_message") or self._default_options.get("system_message")
+        if system_message:
+            config["system_message"] = system_message
 
         if self._tools:
             config["tools"] = self._prepare_tools(self._tools)
 
-        if self._permission_handler:
-            config["on_permission_request"] = self._permission_handler
+        permission_handler = opts.get("on_permission_request") or self._permission_handler
+        if permission_handler:
+            config["on_permission_request"] = permission_handler
 
-        if self._mcp_servers:
-            config["mcp_servers"] = self._mcp_servers
+        mcp_servers = opts.get("mcp_servers") or self._mcp_servers
+        if mcp_servers:
+            config["mcp_servers"] = mcp_servers
 
         return await self._client.create_session(config)
 

python/packages/github_copilot/tests/test_github_copilot_agent.py

@@ -135,12 +135,52 @@ def my_tool(arg: str) -> str:
         agent = GitHubCopilotAgent(tools=[my_tool])
         assert len(agent._tools) == 1  # type: ignore
 
-    def test_init_with_instructions(self) -> None:
-        """Test initialization with custom instructions."""
+    def test_init_with_instructions_parameter(self) -> None:
+        """Test initialization with instructions parameter."""
+        agent = GitHubCopilotAgent(instructions="You are a helpful assistant.")
+        assert agent._default_options.get("system_message") == {  # type: ignore
+            "mode": "append",
+            "content": "You are a helpful assistant.",
+        }
+
+    def test_init_with_system_message_in_default_options(self) -> None:
+        """Test initialization with system_message object in default_options."""
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            default_options={"system_message": {"mode": "append", "content": "You are a helpful assistant."}}
+        )
+        assert agent._default_options.get("system_message") == {  # type: ignore
+            "mode": "append",
+            "content": "You are a helpful assistant.",
+        }
+
+    def test_init_with_system_message_replace_mode(self) -> None:
+        """Test initialization with system_message in replace mode."""
+        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+            default_options={"system_message": {"mode": "replace", "content": "Custom system prompt."}}
+        )
+        assert agent._default_options.get("system_message") == {  # type: ignore
+            "mode": "replace",
+            "content": "Custom system prompt.",
+        }
+
+    def test_instructions_parameter_takes_precedence_for_content(self) -> None:
+        """Test that direct instructions parameter takes precedence for content but preserves mode."""
         agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
-            default_options={"instructions": "You are a helpful assistant."}
+            instructions="Direct instructions",
+            default_options={"system_message": {"mode": "replace", "content": "Options system_message"}},
         )
-        assert agent._instructions == "You are a helpful assistant."  # type: ignore
+        assert agent._default_options.get("system_message") == {  # type: ignore
+            "mode": "replace",
+            "content": "Direct instructions",
+        }
+
+    def test_instructions_parameter_defaults_to_append_mode(self) -> None:
+        """Test that instructions parameter defaults to append mode when no system_message provided."""
+        agent = GitHubCopilotAgent(instructions="Direct instructions")
+        assert agent._default_options.get("system_message") == {  # type: ignore
+            "mode": "append",
+            "content": "Direct instructions",
+        }
 
 
 class TestGitHubCopilotAgentLifecycle:
@@ -462,10 +502,10 @@ async def test_session_config_includes_instructions(
         mock_client: MagicMock,
         mock_session: MagicMock,
     ) -> None:
-        """Test that session config includes instructions."""
-        agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
+        """Test that session config includes instructions from direct parameter."""
+        agent = GitHubCopilotAgent(
+            instructions="You are a helpful assistant.",
             client=mock_client,
-            default_options={"instructions": "You are a helpful assistant."},
         )
         await agent.start()
 
@@ -476,6 +516,31 @@ async def test_session_config_includes_instructions(
         assert config["system_message"]["mode"] == "append"
         assert config["system_message"]["content"] == "You are a helpful assistant."
 
+    async def test_runtime_options_take_precedence_over_default(
+        self,
+        mock_client: MagicMock,
+        mock_session: MagicMock,
+    ) -> None:
+        """Test that runtime options from run() take precedence over default_options."""
+        agent = GitHubCopilotAgent(
+            instructions="Default instructions",
+            client=mock_client,
+        )
+        await agent.start()
+
+        runtime_options: GitHubCopilotOptions = {
+            "system_message": {"mode": "replace", "content": "Runtime instructions"}
+        }
+        await agent._get_or_create_session(  # type: ignore
+            AgentThread(),
+            runtime_options=runtime_options,
+        )
+
+        call_args = mock_client.create_session.call_args
+        config = call_args[0][0]
+        assert config["system_message"]["mode"] == "replace"
+        assert config["system_message"]["content"] == "Runtime instructions"
+
     async def test_session_config_includes_streaming_flag(
         self,
         mock_client: MagicMock,

python/samples/getting_started/agents/github_copilot/github_copilot_basic.py

@@ -18,7 +18,7 @@
 from typing import Annotated
 
 from agent_framework import tool
-from agent_framework.github import GitHubCopilotAgent, GitHubCopilotOptions
+from agent_framework.github import GitHubCopilotAgent
 from pydantic import Field
 
 
@@ -36,8 +36,8 @@ async def non_streaming_example() -> None:
     """Example of non-streaming response (get the complete result at once)."""
     print("=== Non-streaming Response Example ===")
 
-    agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
-        default_options={"instructions": "You are a helpful weather agent."},
+    agent = GitHubCopilotAgent(
+        instructions="You are a helpful weather agent.",
         tools=[get_weather],
     )
 
@@ -52,8 +52,8 @@ async def streaming_example() -> None:
     """Example of streaming response (get results as they are generated)."""
     print("=== Streaming Response Example ===")
 
-    agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
-        default_options={"instructions": "You are a helpful weather agent."},
+    agent = GitHubCopilotAgent(
+        instructions="You are a helpful weather agent.",
         tools=[get_weather],
     )
 
@@ -67,11 +67,46 @@ async def streaming_example() -> None:
         print("\n")
 
 
+async def runtime_options_example() -> None:
+    """Example of overriding system message at runtime."""
+    print("=== Runtime Options Example ===")
+
+    agent = GitHubCopilotAgent(
+        instructions="Always respond in exactly 3 words.",
+        tools=[get_weather],
+    )
+
+    async with agent:
+        query = "What's the weather like in Paris?"
+
+        # First call uses default instructions (3 words response)
+        print("Using default instructions (3 words):")
+        print(f"User: {query}")
+        result1 = await agent.run(query)
+        print(f"Agent: {result1}\n")
+
+        # Second call overrides with runtime system_message in replace mode
+        print("Using runtime system_message with replace mode (detailed response):")
+        print(f"User: {query}")
+        result2 = await agent.run(
+            query,
+            options={
+                "system_message": {
+                    "mode": "replace",
+                    "content": "You are a weather expert. Provide detailed weather information "
+                    "with temperature, and recommendations.",
+                }
+            },
+        )
+        print(f"Agent: {result2}\n")
+
+
 async def main() -> None:
     print("=== Basic GitHub Copilot Agent Example ===")
 
     await non_streaming_example()
     await streaming_example()
+    await runtime_options_example()
 
 
 if __name__ == "__main__":

python/samples/getting_started/agents/github_copilot/github_copilot_with_file_operations.py

@@ -14,7 +14,7 @@
 
 import asyncio
 
-from agent_framework.github import GitHubCopilotAgent, GitHubCopilotOptions
+from agent_framework.github import GitHubCopilotAgent
 from copilot.types import PermissionRequest, PermissionRequestResult
 
 
@@ -35,11 +35,9 @@ def prompt_permission(request: PermissionRequest, context: dict[str, str]) -> Pe
 async def main() -> None:
     print("=== GitHub Copilot Agent with File Operation Permissions ===\n")
 
-    agent: GitHubCopilotAgent[GitHubCopilotOptions] = GitHubCopilotAgent(
-        default_options={
-            "instructions": "You are a helpful assistant that can read and write files.",
-            "on_permission_request": prompt_permission,
-        },
+    agent = GitHubCopilotAgent(
+        instructions="You are a helpful assistant that can read and write files.",
+        default_options={"on_permission_request": prompt_permission},
     )
 
     async with agent: