feat(linear): resolve API token via AgentCore Identity (Phase 2.0a)

agent/pyproject.toml

@@ -5,6 +5,7 @@ description = "Background coding agent — runs tasks in isolated cloud environm
 requires-python = ">=3.13"
 dependencies = [
     "boto3==1.43.9", #https://pypi.org/project/boto3/
+    "bedrock-agentcore==1.9.1", #https://pypi.org/project/bedrock-agentcore/
     "claude-agent-sdk==0.2.82", #https://github.com/anthropics/claude-agent-sdk-python/releases/tag/v0.2.82
     "requests==2.34.2", #https://pypi.org/project/requests/
     "fastapi==0.136.1", #https://pypi.org/project/fastapi/

agent/src/config.py

@@ -39,54 +39,101 @@ def resolve_github_token() -> str:
 
 
 def resolve_linear_api_token() -> str:
-    """Resolve the Linear personal API token from Secrets Manager or env.
+    """Resolve the Linear personal API token via AgentCore Identity.
 
-    Mirrors ``resolve_github_token``: in deployed mode
-    ``LINEAR_API_TOKEN_SECRET_ARN`` is set and the token is fetched once
-    and cached in ``LINEAR_API_TOKEN``. For local development, falls back
-    to ``LINEAR_API_TOKEN`` directly.
+    In deployed mode, ``LINEAR_API_KEY_PROVIDER_NAME`` names a credential
+    provider in AgentCore Identity (the token vault). The agent runtime
+    auto-injects a workload access token into ``BedrockAgentCoreContext``;
+    we exchange that for the API key value and cache it in
+    ``LINEAR_API_TOKEN`` so downstream consumers (the Linear MCP's
+    ``${LINEAR_API_TOKEN}`` placeholder in ``.mcp.json`` and
+    ``linear_reactions.py``'s GraphQL Authorization header) keep working
+    unchanged.
 
-    Returns an empty string if the secret is absent or empty — the agent-side
+    For local development, falls back to a pre-set ``LINEAR_API_TOKEN``
+    env var so the agent can run outside AgentCore Runtime.
+
+    Returns an empty string if the credential is absent — the agent-side
     MCP config then renders with an unresolved ``${LINEAR_API_TOKEN}`` env
     placeholder, and the Linear MCP will reject the request (fail-closed).
     This function is only called when ``channel_source == 'linear'``.
+
+    Phase 2.0a: replaces the prior Secrets Manager path. Phase 2.0b will
+    swap this function entirely for the ``@requires_access_token`` OAuth
+    decorator pattern; this imperative shape exists because API keys
+    don't need refresh and the MCP config expects a static token.
     """
     cached = os.environ.get("LINEAR_API_TOKEN", "")
     if cached:
         return cached
-    secret_arn = os.environ.get("LINEAR_API_TOKEN_SECRET_ARN")
-    if not secret_arn:
+
+    provider_name = os.environ.get("LINEAR_API_KEY_PROVIDER_NAME")
+    if not provider_name:
         return ""
+
+    region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
+    if not region:
+        log("WARN", "resolve_linear_api_token: AWS_REGION not set; cannot resolve API key")
+        return ""
+
     try:
-        import boto3
+        import asyncio
+
+        from bedrock_agentcore.runtime import BedrockAgentCoreContext
+        from bedrock_agentcore.services.identity import IdentityClient
         from botocore.exceptions import BotoCoreError, ClientError
     except ImportError as e:
-        # boto3 missing from the container image — degrade gracefully rather
-        # than hard-crashing the agent. The Linear MCP will fail on first
-        # call with a clear auth error.
-        log("WARN", f"resolve_linear_api_token: boto3 unavailable ({e}); skipping")
+        # bedrock_agentcore SDK missing from the container image — degrade
+        # gracefully rather than hard-crashing the agent. The Linear MCP
+        # will fail on first call with a clear auth error.
+        log("WARN", f"resolve_linear_api_token: bedrock_agentcore unavailable ({e}); skipping")
         return ""
 
     try:
-        region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
-        client = boto3.client("secretsmanager", region_name=region)
-        resp = client.get_secret_value(SecretId=secret_arn)
-        token = resp.get("SecretString", "") or ""
+        workload_token = BedrockAgentCoreContext.get_workload_access_token()
+        if workload_token is None:
+            # Outside the AgentCore Runtime container (e.g. local dev). The
+            # SDK's `_set_up_local_auth` fallback writes `.agentcore.json`
+            # which doesn't fit our flow; bail out and let the caller see
+            # an empty token so the MCP config fails closed.
+            log(
+                "WARN",
+                "resolve_linear_api_token: workload access token not in context "
+                "(agent must run inside AgentCore Runtime, or set LINEAR_API_TOKEN "
+                "directly for local dev)",
+            )
+            return ""
+
+        client = IdentityClient(region=region)
+        token = (
+            asyncio.run(
+                client.get_api_key(
+                    provider_name=provider_name,
+                    agent_identity_token=workload_token,
+                )
+            )
+            or ""
+        )
         if token:
             os.environ["LINEAR_API_TOKEN"] = token
         return token
     except ClientError as e:
-        # Narrowed from a broader `except` per #63 review — broader catches
-        # hid genuine bugs in the Secrets Manager call shape. AccessDenied
-        # is logged at ERROR because it's a persistent IAM misconfig that
-        # should page someone, not a transient blip.
+        # Narrowed from a broader `except` per #63 review. AccessDenied is
+        # logged at ERROR because it's a persistent IAM misconfig (likely
+        # the runtime role missing bedrock-agentcore:GetResourceApiKey or
+        # GetWorkloadAccessToken) that should page someone, not a transient
+        # blip. ResourceNotFound (provider name unknown) is also persistent
+        # — same severity. Other ClientErrors are likely transient (throttle,
+        # network blip) and stay at WARN.
         code = e.response.get("Error", {}).get("Code", "")
-        severity = "ERROR" if code == "AccessDeniedException" else "WARN"
+        severity = (
+            "ERROR" if code in ("AccessDeniedException", "ResourceNotFoundException") else "WARN"
+        )
         log(severity, f"resolve_linear_api_token failed: {type(e).__name__}: {e}")
         return ""
     except BotoCoreError as e:
-        # Never let a Secrets Manager outage crash the agent. The Linear MCP
-        # will simply fail on first call with a clear auth error.
+        # Never let an Identity outage crash the agent. The Linear MCP will
+        # fail on first call with a clear auth error.
         log("WARN", f"resolve_linear_api_token failed: {type(e).__name__}: {e}")
         return ""
 

agent/src/server.py

@@ -209,6 +209,38 @@ async def lifespan(_application: FastAPI):
 app = FastAPI(title="Background Agent", version="1.0.0", lifespan=lifespan)
 
 
+def _extract_workload_access_token(request: Request) -> str:
+    """Read AgentCore's workload access token off the inbound request.
+
+    AgentCore Runtime delivers the token on `/invocations` requests under
+    one of two header spellings (both observed 2026-05-18 on a single
+    request via diagnostic logging in us-east-1):
+      1. ``WorkloadAccessToken`` — the SDK's documented header in
+         ``bedrock_agentcore.runtime.models::ACCESS_TOKEN_HEADER``.
+      2. ``x-amzn-bedrock-agentcore-runtime-workload-accesstoken`` —
+         undocumented but present on the wire; included for forward
+         compatibility.
+
+    The token must be propagated explicitly into the pipeline thread (see
+    ``_run_task_background``) because Python ``ContextVar`` is per-thread,
+    not per-request — the SDK's bundled ``_build_request_context``
+    middleware sets it in the request handler's async context, but our
+    pipeline runs in a separate ``threading.Thread`` spawned by
+    ``_spawn_background``. The new thread sees a fresh empty ContextVar
+    unless we re-set it on entry.
+
+    See aws/bedrock-agentcore-sdk-python#219 for the upstream tracking
+    issue (per-thread ContextVar) and the workaround pattern in
+    ``awslabs/agentcore-samples`` 07-Outbound_Auth_3LO_ECS_Fargate.
+    """
+    return (
+        request.headers.get("WorkloadAccessToken")
+        or request.headers.get("x-amzn-bedrock-agentcore-runtime-workload-accesstoken")
+        or request.headers.get("x-amzn-bedrock-agentcore-workload-access-token")
+        or ""
+    )
+
+
 class InvocationRequest(BaseModel):
     input: dict[str, Any]
 
@@ -277,10 +309,24 @@ def _run_task_background(
     channel_metadata: dict[str, str] | None = None,
     trace: bool = False,
     user_id: str = "",
+    workload_access_token: str = "",
 ) -> None:
     """Run the agent task in a background thread."""
     global _background_pipeline_failed
 
+    # Re-establish the AgentCore workload-token ContextVar in this thread.
+    # Python ContextVar storage is per-thread, so the request-handler thread's
+    # context (where BedrockAgentCoreApp's _build_request_context would normally
+    # set this) doesn't propagate to here. Without this re-set,
+    # IdentityClient.get_api_key() callers like resolve_linear_api_token()
+    # short-circuit on a None workload token even when the platform delivered
+    # one. See aws/bedrock-agentcore-sdk-python#219 for the upstream design
+    # constraint that motivates this manual propagation.
+    if workload_access_token:
+        from bedrock_agentcore.runtime.context import BedrockAgentCoreContext
+
+        BedrockAgentCoreContext.set_workload_access_token(workload_access_token)
+
     _debug_cw(
         f"_run_task_background ENTERED task_id={task_id!r} "
         f"thread={threading.current_thread().name!r}",
@@ -399,6 +445,13 @@ def _extract_invocation_params(inp: dict, request: Request) -> dict:
 
     session_id = request.headers.get("x-amzn-bedrock-agentcore-runtime-session-id", "")
 
+    # AgentCore-injected workload access token (see _extract_workload_access_token
+    # for full rationale). Threaded into _run_task_background so the pipeline
+    # thread can call BedrockAgentCoreContext.set_workload_access_token() on entry
+    # — without that the IdentityClient.get_api_key path used by
+    # resolve_linear_api_token() returns None.
+    workload_access_token = _extract_workload_access_token(request)
+
     return {
         "repo_url": repo_url,
         "task_description": task_description,
@@ -422,6 +475,7 @@ def _extract_invocation_params(inp: dict, request: Request) -> dict:
         "channel_metadata": channel_metadata,
         "trace": trace,
         "user_id": user_id,
+        "workload_access_token": workload_access_token,
     }