feat: external MCP connectors via AgentCore Identity

.gitignore

@@ -90,6 +90,7 @@ tmp/
 
 # Claude - exclude personal settings
 .claude/settings.local.json
+.claude/scheduled_tasks.lock
 
 # OS generated files
 ehthumbs.db

CLAUDE.MD

@@ -48,6 +48,7 @@ npx cdk deploy --all
 | `message_stop` | End of message |
 | `tool_use` / `tool_result` | Tool invocation and result |
 | `stream_error` | Conversational error |
+| `oauth_required` | External MCP tool needs user consent — payload `{providerId, authorizationUrl}`, one event per provider emitted after `message_stop` |
 | `done` | Stream complete |
 
 ## Multi-Protocol Tool Architecture

backend/run-app-api.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")/src/apis/app_api"
+exec uv run python main.py

backend/run-inference-api.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")/src/apis/inference_api"
+exec uv run python main.py

backend/src/agents/main_agent/base_agent.py

@@ -8,12 +8,13 @@
 
 import logging
 from abc import ABC, abstractmethod
-from typing import AsyncGenerator, List, Optional
+from typing import Any, AsyncGenerator, Dict, List, Optional
 
 from agents.main_agent.core import ModelConfig, SystemPromptBuilder, AgentFactory
 from agents.main_agent.session import SessionFactory
 from agents.main_agent.session.hooks import (
     StopHook,
+    OAuthConsentHook,
     EmailApprovalHook,
     ExternalWriteApprovalHook,
     DangerousToolApprovalHook,
@@ -88,6 +89,19 @@ def __init__(
             model_id=model_id, temperature=temperature, caching_enabled=caching_enabled, provider=provider, max_tokens=max_tokens
         )
 
+        # Frozen snapshot of agent-construction params, used when the turn
+        # pauses on OAuth consent so the resume request can rebuild this exact
+        # agent shape without depending on the in-process agent cache.
+        # ``system_prompt`` is captured below after the prompt builder resolves.
+        self._construction_snapshot: dict = {
+            "enabled_tools": enabled_tools,
+            "model_id": model_id,
+            "provider": provider,
+            "temperature": temperature,
+            "caching_enabled": caching_enabled,
+            "max_tokens": max_tokens,
+        }
+
         # Load retry configuration from environment variables
         from agents.main_agent.core.model_config import RetryConfig
         self.model_config.retry_config = RetryConfig.from_env()
@@ -100,6 +114,10 @@ def __init__(
             self.prompt_builder = SystemPromptBuilder()
             self.system_prompt = self.prompt_builder.build(include_date=True)
 
+        # Capture the resolved system prompt — what we'd need to pass back to
+        # ``get_agent`` to land on the same cache key on resume.
+        self._construction_snapshot["system_prompt"] = self.system_prompt
+
         # Initialize tool registry and filter
         self.tool_registry = create_default_registry()
         self.tool_filter = ToolFilter(self.tool_registry)
@@ -131,9 +149,21 @@ def _create_agent(self) -> None:
 
     @abstractmethod
     async def stream_async(
-        self, message: str, session_id: Optional[str] = None, files: Optional[List] = None, citations: Optional[List] = None, original_message: Optional[str] = None
+        self,
+        message: str,
+        session_id: Optional[str] = None,
+        files: Optional[List] = None,
+        citations: Optional[List] = None,
+        original_message: Optional[str] = None,
+        interrupt_responses: Optional[List[Dict[str, Any]]] = None,
     ) -> AsyncGenerator[str, None]:
-        """Stream agent responses. Subclasses must implement."""
+        """Stream agent responses. Subclasses must implement.
+
+        When `interrupt_responses` is provided, the call resumes a paused
+        agent turn (Strands interrupt protocol) instead of starting a new
+        one. In that case `message`/`files` are ignored — the original turn
+        already has the user's prompt in its context.
+        """
         ...
 
     def _register_external_mcp_tools(self) -> None:
@@ -187,6 +217,8 @@ def _create_hooks(self) -> List:
 
         Includes:
         - StopHook: Always enabled, cancels tool execution on user stop
+        - OAuthConsentHook: Pauses the agent (Strands interrupt) when an
+          OAuth-gated MCP tool is about to run without a cached token
         - Approval hooks: Gate dangerous operations for user confirmation
 
         Returns:
@@ -197,13 +229,69 @@ def _create_hooks(self) -> List:
         # Always-on: session cancellation
         hooks.append(StopHook(self.session_manager))
 
+        # OAuth consent gate for external MCP tools. Registered unconditionally;
+        # the hook is a no-op for tools that don't have a registered provider.
+        hooks.append(self._build_oauth_consent_hook())
+
         # Approval gates for dangerous operations
         hooks.append(EmailApprovalHook())
         hooks.append(ExternalWriteApprovalHook())
         hooks.append(DangerousToolApprovalHook())
 
         return hooks
 
+    def _build_oauth_consent_hook(self) -> OAuthConsentHook:
+        """Construct the OAuth consent hook with closures over the MCP
+        integration and provider repository so it stays decoupled from them.
+        """
+        from agents.main_agent.integrations.external_mcp_client import (
+            get_external_mcp_integration,
+        )
+        from strands.tools.mcp import MCPAgentTool
+
+        integration = get_external_mcp_integration()
+
+        def provider_lookup(selected_tool: object) -> Optional[str]:
+            if not isinstance(selected_tool, MCPAgentTool):
+                return None
+            return integration.provider_for_client(selected_tool.mcp_client)
+
+        async def scopes_lookup(provider_id: str) -> List[str]:
+            from apis.shared.oauth.provider_repository import get_provider_repository
+
+            provider = await get_provider_repository().get_provider(provider_id)
+            return provider.scopes if provider else []
+
+        async def provider_type_lookup(provider_id: str) -> Optional[str]:
+            # AgentCore Identity needs vendor-specific OAuth params
+            # forwarded via `customParameters` (e.g. Google's
+            # `access_type=offline` for refresh tokens). The hook reads
+            # this to forward those.
+            from apis.shared.oauth.provider_repository import get_provider_repository
+
+            provider = await get_provider_repository().get_provider(provider_id)
+            return provider.provider_type.value if provider else None
+
+        async def custom_parameters_lookup(
+            provider_id: str,
+        ) -> Optional[dict[str, str]]:
+            # Admin-supplied OAuth extras (e.g. `hd=mycorp.com` for
+            # Google Workspace domain restriction). Merged with the
+            # vendor baseline by `custom_parameters_for`; baseline wins
+            # on conflict.
+            from apis.shared.oauth.provider_repository import get_provider_repository
+
+            provider = await get_provider_repository().get_provider(provider_id)
+            return provider.custom_parameters if provider else None
+
+        return OAuthConsentHook(
+            user_id=self.user_id,
+            provider_lookup=provider_lookup,
+            scopes_lookup=scopes_lookup,
+            provider_type_lookup=provider_type_lookup,
+            custom_parameters_lookup=custom_parameters_lookup,
+        )
+
     def _build_filtered_tools(self) -> List:
         """
         Filter tools and load gateway/external MCP clients.
@@ -226,22 +314,34 @@ def _build_filtered_tools(self) -> List:
         if external_mcp_tool_ids:
             import asyncio
 
+            from bedrock_agentcore.runtime import BedrockAgentCoreContext
+
             from agents.main_agent.integrations.external_mcp_client import get_external_mcp_integration
 
+            # Capture request-scoped context values before crossing the thread
+            # boundary below. ContextVars do not propagate into the executor's
+            # fresh event loop, so anything we need there must be passed as args.
+            oauth2_callback_url = BedrockAgentCoreContext.get_oauth2_callback_url()
+            workload_access_token = BedrockAgentCoreContext.get_workload_access_token()
+
             external_integration = get_external_mcp_integration()
             loop = asyncio.get_event_loop()
             if loop.is_running():
                 import concurrent.futures
 
-                with concurrent.futures.ThreadPoolExecutor() as executor:
-                    future = executor.submit(
-                        asyncio.run,
-                        external_integration.load_external_tools(
-                            external_mcp_tool_ids,
-                            user_id=self.user_id,
-                            auth_token=self.auth_token,
-                        ),
+                async def _load_with_context():
+                    if oauth2_callback_url:
+                        BedrockAgentCoreContext.set_oauth2_callback_url(oauth2_callback_url)
+                    if workload_access_token:
+                        BedrockAgentCoreContext.set_workload_access_token(workload_access_token)
+                    return await external_integration.load_external_tools(
+                        external_mcp_tool_ids,
+                        user_id=self.user_id,
+                        auth_token=self.auth_token,
                     )
+
+                with concurrent.futures.ThreadPoolExecutor() as executor:
+                    future = executor.submit(asyncio.run, _load_with_context())
                     external_clients = future.result()
             else:
                 external_clients = loop.run_until_complete(

backend/src/agents/main_agent/chat_agent.py

@@ -6,7 +6,7 @@
 """
 
 import logging
-from typing import AsyncGenerator, List, Optional
+from typing import Any, AsyncGenerator, Dict, List, Optional
 
 from agents.main_agent.base_agent import BaseAgent
 from agents.main_agent.core import AgentFactory
@@ -43,25 +43,43 @@ def _create_agent(self) -> None:
             raise
 
     async def stream_async(
-        self, message: str, session_id: Optional[str] = None, files: Optional[List] = None, citations: Optional[List] = None, original_message: Optional[str] = None
+        self,
+        message: str,
+        session_id: Optional[str] = None,
+        files: Optional[List] = None,
+        citations: Optional[List] = None,
+        original_message: Optional[str] = None,
+        interrupt_responses: Optional[List[Dict[str, Any]]] = None,
     ) -> AsyncGenerator[str, None]:
         """
         Stream agent responses.
 
         Args:
-            message: User message text
+            message: User message text. Ignored when resuming via
+                `interrupt_responses` — the paused turn already has the
+                original prompt in `_interrupt_state`.
             session_id: Session identifier (defaults to instance session_id)
             files: Optional list of FileContent objects (with base64 bytes)
             citations: Optional list of citation dicts from RAG retrieval
             original_message: Original user message before RAG augmentation
+            interrupt_responses: When set, resume a paused agent turn by
+                passing this list as the prompt to Strands. Each entry is
+                `{"interruptResponse": {"interruptId": str, "response": Any}}`.
 
         Yields:
             str: SSE formatted events
         """
         if not self.agent:
             self._create_agent()
 
-        prompt = self.multimodal_builder.build_prompt(message, files)
+        if interrupt_responses:
+            # Strands' resume protocol: passing a list of interrupt responses
+            # as the prompt re-enters the loop, populates the matching
+            # interrupts' `.response`, and continues from the paused tool
+            # call. multimodal_builder + files do not apply here.
+            prompt: Any = interrupt_responses
+        else:
+            prompt = self.multimodal_builder.build_prompt(message, files)
 
         async for event in self.stream_coordinator.stream_response(
             agent=self.agent,