feat: AgentCore Runtime improvements — MCP server_instructions, iteration limit, Code Interpreter option, S3 link format

docs/en/DEPLOY_OPTION.md

@@ -727,6 +727,8 @@ This is a use case for integrating with agents created in AgentCore. (Experiment
 Enabling `createGenericAgentCoreRuntime` will deploy the default AgentCore Runtime.
 By default, it is deployed to `modelRegion`, but you can override it by specifying `agentCoreRegion`.
 
+Setting `agentCoreCodeInterpreterEnabled` to `true` enables Code Interpreter in AgentCore Chat. This allows the agent to execute Python code and shell commands within a sandboxed environment. Note that the Code Interpreter sandbox is isolated from the host filesystem — files created in the sandbox are not accessible to MCP tools, and vice versa.
+
 The default agents available in AgentCore can use MCP servers defined in [generic/mcp.json](packages/cdk/lambda-python/generic-agent-core-runtime/mcp-configs/generic/mcp.json).
 
 The MCP servers defined by default are AWS-related MCP servers and MCP servers related to current time.
@@ -766,6 +768,7 @@ To enable the AgentCore use case, the `docker` command must be executable.
 const envs: Record<string, Partial<StackInput>> = {
   dev: {
     createGenericAgentCoreRuntime: true,
+    agentCoreCodeInterpreterEnabled: true,
     agentCoreRegion: 'us-west-2',
     agentCoreGatewayArns: [
       'arn:aws:bedrock-agentcore:us-west-2:<account>:gateway/<gateway-id>',
@@ -788,6 +791,7 @@ const envs: Record<string, Partial<StackInput>> = {
 {
   "context": {
     "createGenericAgentCoreRuntime": true,
+    "agentCoreCodeInterpreterEnabled": true,
     "agentCoreRegion": "us-west-2",
     "agentCoreGatewayArns": [
       "arn:aws:bedrock-agentcore:us-west-2:<account>:gateway/<gateway-id>"

docs/ja/DEPLOY_OPTION.md

@@ -746,6 +746,8 @@ AgentCore で作成したエージェントと連携するユースケースで
 `createGenericAgentCoreRuntime` を有効化するとデフォルトの AgentCore Runtime がデプロイされます。
 デフォルトでは `modelRegion` にデプロイされますが、`agentCoreRegion` を指定し上書きすることが可能です。
 
+`agentCoreCodeInterpreterEnabled` を `true` にすると、AgentCore チャットで Code Interpreter が有効になります。エージェントがサンドボックス環境内で Python コードやシェルコマンドを実行できるようになります。なお、Code Interpreter のサンドボックスはホストのファイルシステムとは分離されており、サンドボックス内で作成したファイルは MCP ツールからアクセスできません（逆も同様です）。
+
 AgentCore で使用できるデフォルトのエージェントは、[generic/mcp.json](packages/cdk/lambda-python/generic-agent-core-runtime/mcp-configs/generic/mcp.json) で定義する MCP サーバーを利用することができます。
 
 デフォルトで定義されている MCP サーバーは、AWS に関連する MCP サーバー及び、現在時刻に関連する MCP サーバーです。
@@ -785,6 +787,7 @@ AgentCore ユースケースを有効化するためには、`docker` コマン
 const envs: Record<string, Partial<StackInput>> = {
   dev: {
     createGenericAgentCoreRuntime: true,
+    agentCoreCodeInterpreterEnabled: true,
     agentCoreRegion: 'us-west-2',
     agentCoreGatewayArns: [
       'arn:aws:bedrock-agentcore:us-west-2:<account>:gateway/<gateway-id>',
@@ -807,6 +810,7 @@ const envs: Record<string, Partial<StackInput>> = {
 {
   "context": {
     "createGenericAgentCoreRuntime": true,
+    "agentCoreCodeInterpreterEnabled": true,
     "agentCoreRegion": "us-west-2",
     "agentCoreGatewayArns": [
       "arn:aws:bedrock-agentcore:us-west-2:<account>:gateway/<gateway-id>"

packages/cdk/cdk.json

@@ -68,6 +68,7 @@
     "flows": [],
     "agentBuilderEnabled": false,
     "createGenericAgentCoreRuntime": false,
+    "agentCoreCodeInterpreterEnabled": false,
     "agentCoreRegion": null,
     "agentCoreExternalRuntimes": [],
     "agentCoreGatewayArns": [],

packages/cdk/lambda-python/generic-agent-core-runtime/src/agent.py

@@ -75,6 +75,12 @@ async def process_request_streaming(
             tools = self.tool_manager.get_tools_with_options(code_execution_enabled=code_execution_enabled, mcp_servers=mcp_servers)
             logger.info(f"Loaded {len(tools)} tools (code execution: {code_execution_enabled})")
 
+            # Inject MCP Server Instructions into system prompt
+            mcp_instructions = self.tool_manager.get_mcp_instructions()
+            if mcp_instructions:
+                combined_system_prompt += f"\n\n## MCP Server Instructions\n\n{mcp_instructions}"
+                logger.info(f"Injected MCP Server Instructions ({len(mcp_instructions)} chars) into system prompt")
+
             # Log agent info
             if agent_id:
                 logger.debug(f"Processing agent: {agent_id}")

packages/cdk/lambda-python/generic-agent-core-runtime/src/config.py

@@ -15,13 +15,14 @@
 
 WORKSPACE_DIR = "/tmp/ws"
 
-DEFAULT_MAX_ITERATIONS = 20
+DEFAULT_MAX_ITERATIONS = 100
 
 FIXED_SYSTEM_PROMPT = f"""## About File Output
 - You are running on AWS Bedrock AgentCore. Therefore, when writing files, always write them under `{WORKSPACE_DIR}`.
 - Similarly, if you need a workspace, please use the `{WORKSPACE_DIR}` directory. Do not ask the user about their current workspace. It's always `{WORKSPACE_DIR}`.
 - Also, users cannot directly access files written under `{WORKSPACE_DIR}`. So when submitting these files to users, *always upload them to S3 using the `upload_file_to_s3_and_retrieve_s3_url` tool and provide the S3 URL*. The S3 URL must be included in the final output.
 - If the output file is an image file, the S3 URL output must be in Markdown format.
+- For all other file types (e.g. .pptx, .csv, .pdf), the S3 URL must also be in Markdown link format: `[filename](S3_URL)`. This enables the UI to generate a download link.
 """
 
 

packages/cdk/lambda-python/generic-agent-core-runtime/src/tools.py

@@ -51,6 +51,7 @@ class ToolManager:
 
     def __init__(self):
         self.mcp_tools = None
+        self.mcp_instructions: list[str] = []
         self.session_id = None
         self.trace_id = None
 
@@ -89,6 +90,13 @@ def load_mcp_tools(self) -> list[Any]:
 
             # Flatten the tools
             self.mcp_tools = sum([c.list_tools_sync() for c in mcp_clients], [])
+
+            # Collect server instructions from MCP servers
+            for client in mcp_clients:
+                if hasattr(client, "server_instructions") and client.server_instructions:
+                    logger.info(f"Collected server instructions ({len(client.server_instructions)} chars)")
+                    self.mcp_instructions.append(client.server_instructions)
+
             logger.info(f"Loaded {len(self.mcp_tools)} MCP tools")
             return self.mcp_tools
 
@@ -131,6 +139,13 @@ def load_mcp_tools_by_names(self, server_names: list[str]) -> list[Any]:
 
             # Flatten the tools
             dynamic_tools = sum([c.list_tools_sync() for c in mcp_clients], [])
+
+            # Collect server instructions from dynamically loaded MCP servers
+            for client in mcp_clients:
+                if hasattr(client, "server_instructions") and client.server_instructions:
+                    logger.info(f"Collected server instructions ({len(client.server_instructions)} chars)")
+                    self.mcp_instructions.append(client.server_instructions)
+
             logger.info(f"Loaded {len(dynamic_tools)} MCP tools from {len(mcp_clients)} servers")
             return dynamic_tools
 
@@ -176,6 +191,123 @@ def upload_file_to_s3_and_retrieve_s3_url(filepath: str) -> str:
 
         return upload_file_to_s3_and_retrieve_s3_url
 
+    def get_file_write_tool(self):
+        """Get the file write tool scoped to WORKSPACE_DIR"""
+
+        @tool
+        def write_file(filepath: str, content: str, mode: str = "create", old_str: str = "", new_str: str = "") -> str:
+            """Write, append, or edit a file under /tmp/ws.
+
+            Args:
+                filepath: Path to the file (must be under /tmp/ws).
+                content: Text content to write (for create/append modes).
+                mode: "create" to create/overwrite, "append" to append, "str_replace" to replace text.
+                old_str: Text to find (required for str_replace mode). Must match exactly once.
+                new_str: Replacement text (for str_replace mode). Empty string to delete.
+            """
+            filepath = os.path.normpath(filepath)
+            if not filepath.startswith(WORKSPACE_DIR):
+                raise ValueError(f"Path must be under {WORKSPACE_DIR}. Got: {filepath}")
+            os.makedirs(os.path.dirname(filepath), exist_ok=True)
+
+            if mode == "str_replace":
+                if not old_str:
+                    raise ValueError("old_str is required for str_replace mode")
+                with open(filepath, encoding="utf-8") as f:
+                    text = f.read()
+                count = text.count(old_str)
+                if count == 0:
+                    raise ValueError(f"old_str not found in {filepath}")
+                if count > 1:
+                    raise ValueError(f"old_str found {count} times in {filepath}. Must be unique.")
+                text = text.replace(old_str, new_str, 1)
+                with open(filepath, "w", encoding="utf-8") as f:
+                    f.write(text)
+                return f"Replaced in {filepath}"
+            else:
+                flag = "a" if mode == "append" else "w"
+                with open(filepath, flag, encoding="utf-8") as f:
+                    f.write(content)
+                return f"Wrote {len(content)} chars to {filepath} (mode={mode})"
+
+        return write_file
+
+    def get_concat_files_tool(self):
+        """Get the file concatenation tool scoped to WORKSPACE_DIR"""
+
+        @tool
+        def concat_files(source_paths: list[str], destination: str) -> str:
+            """Concatenate multiple files into one. All paths must be under /tmp/ws.
+
+            Args:
+                source_paths: List of file paths to concatenate in order.
+                destination: Output file path.
+            """
+            for p in source_paths + [destination]:
+                normed = os.path.normpath(p)
+                if not normed.startswith(WORKSPACE_DIR):
+                    raise ValueError(f"Path must be under {WORKSPACE_DIR}. Got: {p}")
+            with open(os.path.normpath(destination), "w", encoding="utf-8") as out:
+                for p in source_paths:
+                    with open(os.path.normpath(p), encoding="utf-8") as f:
+                        out.write(f.read())
+            return f"Concatenated {len(source_paths)} files into {destination}"
+
+        return concat_files
+
+    def get_web_fetch_tool(self):
+        """Get the web fetch tool"""
+
+        @tool
+        def web_fetch(url: str, max_chars: int = 50000) -> str:
+            """Fetch text content from a URL. Useful for reading web pages, documentation, or API responses.
+
+            Args:
+                url: The URL to fetch.
+                max_chars: Maximum characters to return (default 50000).
+            """
+            import urllib.request
+
+            req = urllib.request.Request(url, headers={"User-Agent": "GenU-AgentCore/1.0"})
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                content_type = resp.headers.get("Content-Type", "")
+                raw = resp.read().decode("utf-8", errors="replace")
+
+            # Strip HTML tags for readability if HTML
+            if "html" in content_type:
+                from html.parser import HTMLParser
+
+                class _TextExtractor(HTMLParser):
+                    def __init__(self):
+                        super().__init__()
+                        self._parts: list[str] = []
+                        self._skip = False
+
+                    def handle_starttag(self, tag, attrs):
+                        if tag in ("script", "style"):
+                            self._skip = True
+
+                    def handle_endtag(self, tag):
+                        if tag in ("script", "style"):
+                            self._skip = False
+
+                    def handle_data(self, data):
+                        if not self._skip:
+                            self._parts.append(data)
+
+                extractor = _TextExtractor()
+                extractor.feed(raw)
+                raw = " ".join(extractor._parts)
+                import re
+
+                raw = re.sub(r"\s+", " ", raw).strip()
+
+            if len(raw) > max_chars:
+                raw = raw[:max_chars] + f"\n\n[Truncated at {max_chars} chars]"
+            return raw
+
+        return web_fetch
+
     def get_code_interpreter_tool(self) -> list[Any]:
         """Get code interpreter tool if available"""
         code_interpreter_tools = []
@@ -192,6 +324,20 @@ def get_code_interpreter_tool(self) -> list[Any]:
 
         return code_interpreter_tools
 
+    def get_mcp_instructions(self) -> str:
+        """Return collected MCP Server Instructions as a single string.
+
+        Server Instructions are provided by MCP servers during initialization
+        to guide the LLM on how to use their tools effectively.
+        See: https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#initialization
+
+        Returns:
+            Combined instructions from all connected MCP servers, or empty string if none.
+        """
+        if not self.mcp_instructions:
+            return ""
+        return "\n\n---\n\n".join(inst.strip() for inst in self.mcp_instructions)
+
     def get_tools_with_options(self, code_execution_enabled: bool = False, mcp_servers=None) -> list[Any]:
         """
         Get tools with optional code execution and MCP servers.
@@ -233,7 +379,13 @@ def get_tools_with_options(self, code_execution_enabled: bool = False, mcp_serve
 
         # Add built-in tools (always included)
         upload_tool = self.get_upload_tool()
+        file_write_tool = self.get_file_write_tool()
+        concat_tool = self.get_concat_files_tool()
+        web_fetch_tool = self.get_web_fetch_tool()
         all_tools.append(upload_tool)
+        all_tools.append(file_write_tool)
+        all_tools.append(concat_tool)
+        all_tools.append(web_fetch_tool)
 
         # Add code interpreter tools if enabled
         code_interpreter_tools = []
@@ -242,6 +394,6 @@ def get_tools_with_options(self, code_execution_enabled: bool = False, mcp_serve
             all_tools.extend(code_interpreter_tools)
 
         # Log final tool count
-        logger.info(f"Total tools loaded: {len(all_tools)} (MCP: {len(mcp_tools)}, Built-in: 1, Code Interpreter: {len(code_interpreter_tools)} - {'enabled' if code_execution_enabled else 'disabled'})")
+        logger.info(f"Total tools loaded: {len(all_tools)} (MCP: {len(mcp_tools)}, Built-in: 4, Code Interpreter: {len(code_interpreter_tools)} - {'enabled' if code_execution_enabled else 'disabled'})")
 
         return all_tools

packages/cdk/lambda-python/generic-agent-core-runtime/tests/test_mcp_instructions.py

@@ -0,0 +1,89 @@
+"""Tests for MCP Server Instructions collection and injection."""
+
+from unittest.mock import MagicMock, patch
+
+from src.tools import ToolManager
+
+
+class TestToolManagerInstructions:
+    """Test ToolManager's MCP Server Instructions collection."""
+
+    def test_initial_state_empty(self):
+        tm = ToolManager()
+        assert tm.mcp_instructions == []
+        assert tm.get_mcp_instructions() == ""
+
+    def test_get_mcp_instructions_single(self):
+        tm = ToolManager()
+        tm.mcp_instructions = ["Use tool A before tool B."]
+        assert tm.get_mcp_instructions() == "Use tool A before tool B."
+
+    def test_get_mcp_instructions_multiple(self):
+        tm = ToolManager()
+        tm.mcp_instructions = [
+            "Server 1: Always call init first.",
+            "Server 2: Rate limit is 10 req/min.",
+        ]
+        result = tm.get_mcp_instructions()
+        assert "Server 1: Always call init first." in result
+        assert "Server 2: Rate limit is 10 req/min." in result
+        assert "\n\n---\n\n" in result
+
+    def test_get_mcp_instructions_strips_whitespace(self):
+        tm = ToolManager()
+        tm.mcp_instructions = ["  padded instructions  \n"]
+        assert tm.get_mcp_instructions() == "padded instructions"
+
+    @patch("src.tools._create_mcp_client")
+    @patch("src.tools.os.environ.get")
+    @patch("src.tools.os.path.exists")
+    @patch("builtins.open")
+    def test_load_mcp_tools_collects_instructions(self, mock_open, mock_exists, mock_env_get, mock_create):
+        """Verify load_mcp_tools collects server_instructions from MCPClients."""
+        mock_env_get.return_value = "/tmp/mcp.json"
+        mock_exists.return_value = True
+
+        import json
+
+        mcp_config = {"mcpServers": {"test-server": {"command": "echo", "args": []}}}
+        mock_open.return_value.__enter__ = lambda s: s
+        mock_open.return_value.__exit__ = MagicMock(return_value=False)
+        mock_open.return_value.read = MagicMock(return_value=json.dumps(mcp_config))
+
+        mock_client = MagicMock()
+        mock_client.list_tools_sync.return_value = []
+        mock_client.server_instructions = "Always call init before generate."
+        mock_create.return_value = ("test-server", mock_client)
+
+        tm = ToolManager()
+        tm.load_mcp_tools()
+
+        assert len(tm.mcp_instructions) == 1
+        assert "Always call init before generate." in tm.mcp_instructions[0]
+
+    @patch("src.tools._create_mcp_client")
+    @patch("src.tools.os.environ.get")
+    @patch("src.tools.os.path.exists")
+    @patch("builtins.open")
+    def test_load_mcp_tools_skips_none_instructions(self, mock_open, mock_exists, mock_env_get, mock_create):
+        """Verify load_mcp_tools skips servers without instructions."""
+        mock_env_get.return_value = "/tmp/mcp.json"
+        mock_exists.return_value = True
+
+        import json
+
+        mcp_config = {"mcpServers": {"no-inst": {"command": "echo", "args": []}}}
+        mock_open.return_value.__enter__ = lambda s: s
+        mock_open.return_value.__exit__ = MagicMock(return_value=False)
+        mock_open.return_value.read = MagicMock(return_value=json.dumps(mcp_config))
+
+        mock_client = MagicMock()
+        mock_client.list_tools_sync.return_value = []
+        mock_client.server_instructions = None
+        mock_create.return_value = ("no-inst", mock_client)
+
+        tm = ToolManager()
+        tm.load_mcp_tools()
+
+        assert len(tm.mcp_instructions) == 0
+        assert tm.get_mcp_instructions() == ""