test: actually add a integ test for agentcore runtime

.github/workflows/python-integ.yml

@@ -0,0 +1,49 @@
+name: Python Integration Tests
+
+
+on:
+  workflow_call:
+  workflow_dispatch:
+  # TODO: enable this back once the workflow runs without issues
+  # push:
+  #   branches: main
+
+jobs:
+  integration:
+    environment: Integ
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.12']
+        # python-version: ['3.10', '3.11', '3.12', '3.13']
+    permissions:
+      id-token: write
+      contents: read
+    env:
+      AGENTCORE_RUNTIME_ARN: ${{ secrets.AgentCoreRuntimeArn }}
+
+    steps:
+      - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@b75a909f75acd358c2196fb9a5f1299a9a8868a4 # v6.7.0
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c # v6.0.0
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install dependencies
+        run: uv sync --frozen --all-extras --dev
+
+      - name: Configure AWS Credentials for Tests
+        uses: aws-actions/configure-aws-credentials@v5
+        with:
+          aws-region: us-west-2
+          role-to-assume: ${{ secrets.IntegTestRoleArn }}
+          role-session-name: aws-mcp-proxy-integ-tests
+
+      - name: Run integration tests
+        run: |
+          uv run pytest -m integ -s

tests/integ/test_bedrock_agentcore_runtime.py

@@ -0,0 +1,97 @@
+"""Test the features about testing connecting to agentcore runtime via the proxy."""
+
+import boto3
+import fastmcp
+import os
+import pytest
+import pytest_asyncio
+from fastmcp.client import StdioTransport
+from typing import TypedDict
+
+
+class AgentCoreRuntimeConfig(TypedDict):
+    """AgentCore endpoint config."""
+
+    endpoint: str
+    region_name: str
+
+
+RUNTIME_ENDPOINT_FMT_STR = 'https://bedrock-agentcore.{region_name}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT'
+
+
+def get_aws_credentials():
+    """Return aws credentials for the session."""
+    credentials = boto3.Session().get_credentials()
+    return {
+        'AWS_ACCESS_KEY_ID': credentials.access_key,
+        'AWS_SECRET_ACCESS_KEY': credentials.secret_key,
+        'AWS_SESSION_TOKEN': credentials.token,
+    }
+
+
+@pytest.fixture(scope='module')
+def agentcore_runtime() -> AgentCoreRuntimeConfig:
+    """Get runtime endpoint from arn in environment variable."""
+    runtime_arn = os.environ.get('AGENTCORE_RUNTIME_ARN')
+    if not runtime_arn:
+        raise RuntimeError('Agent core runtime arn not found')
+    region_name = runtime_arn.split(':')[3]
+    encoded_arn = runtime_arn.replace(':', '%3A').replace('/', '%2F')
+
+    return {
+        'endpoint': RUNTIME_ENDPOINT_FMT_STR.format(
+            region_name=region_name,
+            encoded_arn=encoded_arn,
+        ),
+        'region_name': region_name,
+    }
+
+
+@pytest_asyncio.fixture(loop_scope='module', scope='module')
+async def mcp_client(
+    agentcore_runtime: AgentCoreRuntimeConfig,
+):
+    """Create a mcp client that connects to the remote sigv4 MCP via the proxy."""
+    client = fastmcp.Client(
+        StdioTransport(
+            command='aws-mcp-proxy',
+            args=[
+                agentcore_runtime['endpoint'],
+                '--service',
+                'bedrock-agentcore',
+                '--log-level',
+                'DEBUG',
+            ],
+            env={'AWS_REGION': agentcore_runtime['region_name']} | get_aws_credentials(),
+        ),
+    )
+    async with client:
+        yield client
+
+
+@pytest.mark.asyncio(loop_scope='module')
+async def test_ping(mcp_client: fastmcp.Client):
+    """Test ping."""
+    await mcp_client.ping()
+
+
+@pytest.mark.asyncio(loop_scope='module')
+async def test_list_tools(mcp_client: fastmcp.Client):
+    """Test list tool."""
+    tools = await mcp_client.list_tools()
+    assert tools
+
+
+def test_call_tool():
+    """TODO."""
+    pass
+
+
+def test_handle_elicitation():
+    """TODO."""
+    pass
+
+
+def test_handle_sampling():
+    """TODO."""
+    pass