feat(agents): add instruction field to ManagedAgent

Add an `instruction` field (str or InstructionProvider) to ManagedAgent
and a `canonical_instruction` resolver mirroring LlmAgent. This task only
adds the field and resolver; forwarding it to the Managed Agents API is a
follow-up change.

Co-authored-by: Haran Rajkumar <haranrk@google.com>
PiperOrigin-RevId: 951193016

Author: haranrk

src/google/adk/agents/_managed_agent.py

@@ -51,6 +51,7 @@
 from ..utils.content_utils import to_user_content
 from ..utils.context_utils import Aclosing
 from ..utils.env_utils import is_enterprise_mode_enabled
+from ..utils.instructions_utils import InstructionProvider
 from .base_agent import BaseAgent
 from .context import Context
 from .invocation_context import InvocationContext
@@ -144,6 +145,16 @@ class ManagedAgent(BaseAgent):
   agent_config: Optional[CreateAgentInteractionAgentConfigParam] = None
   """Runtime configuration passed to interactions.create."""
 
+  instruction: Union[str, InstructionProvider] = ''
+  """The system instruction sent to the Managed Agent.
+
+  A plain string may embed ``{var}``, ``{artifact.name}``, or ``{var?}``
+  placeholders that are resolved from session state / artifacts at request time
+  (see ``inject_session_state``). An ``InstructionProvider`` callable is invoked
+  with a ``ReadonlyContext`` and bypasses placeholder injection (it manages
+  state itself). Empty by default, in which case no system instruction is sent.
+  """
+
   tools: list[
       Union[types.Tool, BaseTool, Callable[..., Any], RemoteMcpServer]
   ] = Field(default_factory=list)
@@ -197,6 +208,29 @@ def api_client(self) -> Client:
         )
     return self._api_client
 
+  async def canonical_instruction(
+      self, ctx: ReadonlyContext
+  ) -> tuple[str, bool]:
+    """Resolves ``self.instruction`` for the current context.
+
+    Mirrors ``LlmAgent.canonical_instruction``.
+
+    Args:
+      ctx: The read-only context used to resolve an InstructionProvider.
+
+    Returns:
+      A tuple of (instruction, bypass_state_injection).
+      ``bypass_state_injection``
+      is True when the instruction came from an ``InstructionProvider`` callable
+      (which manages state itself), False for a plain string.
+    """
+    if isinstance(self.instruction, str):
+      return self.instruction, False
+    instruction = self.instruction(ctx)
+    if inspect.isawaitable(instruction):
+      instruction = await instruction
+    return instruction, True
+
   async def _resolve_backend_tools(
       self, ctx: InvocationContext
   ) -> list[ToolParam]:

tests/unittests/agents/test_managed_agent.py

@@ -1044,3 +1044,48 @@ def test_run_async_forwards_mcp_server_param():
   mcp = [t for t in create_kwargs['tools'] if t['type'] == 'mcp_server'][0]
   assert mcp['url'] == 'https://mcp.example.com/mcp'
   assert mcp['headers'] == {'X-Goog-Api-Key': 'k'}
+
+
+def test_canonical_instruction_str():
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction='hello',
+      api_client=_FakeClient(),
+  )
+
+  text, bypass = asyncio.run(agent.canonical_instruction(MagicMock()))
+
+  assert text == 'hello'
+  assert bypass is False
+
+
+def test_canonical_instruction_sync_provider():
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction=lambda ctx: 'from provider',
+      api_client=_FakeClient(),
+  )
+
+  text, bypass = asyncio.run(agent.canonical_instruction(MagicMock()))
+
+  assert text == 'from provider'
+  assert bypass is True
+
+
+def test_canonical_instruction_async_provider():
+  async def provider(ctx):
+    return 'async provider'
+
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction=provider,
+      api_client=_FakeClient(),
+  )
+
+  text, bypass = asyncio.run(agent.canonical_instruction(MagicMock()))
+
+  assert text == 'async provider'
+  assert bypass is True