### Start Example Server

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/TESTING_GUIDE.md

Navigate to the async integration tests directory and start the example server using the provided Python script. Ensure you are in the correct directory.

```bash
cd tests_integ/async

python async_status_example.py
```

--------------------------------

### Local Development Setup

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Clone the repository, install dependencies, and activate the virtual environment for local development.

```bash
git clone https://github.com/aws/bedrock-agentcore-sdk-python.git
cd bedrock-agentcore-sdk-python
uv sync
source .venv/bin/activate
```

--------------------------------

### Install Bedrock AgentCore SDK

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/README.md

Install the SDK using pip. This command is used for setting up the library in your Python environment.

```bash
pip install bedrock-agentcore
```

--------------------------------

### Clone Repository and Install Dependencies

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/CONTRIBUTING.md

Clone the repository and set up the development environment by creating a virtual environment and installing project dependencies using uv.

```bash
git clone https://github.com/aws/bedrock-agentcore-sdk-python.git
cd bedrock-agentcore-sdk-python

uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv sync

pre-commit install
```

--------------------------------

### Install AgentCore CLI - Bash

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

Install the AgentCore CLI globally using npm for agent creation and deployment.

```bash
npm i -g @aws/agentcore
```

--------------------------------

### Install AgentCore with AG-UI Support - Bash

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

Install the bedrock-agentcore package with AG-UI capabilities using pip.

```bash
pip install "bedrock-agentcore[ag-ui]"
```

--------------------------------

### Install Strands AgentCore Payments Plugin

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

Install the plugin using pip. For local development, clone the repository and set up the environment.

```bash
pip install 'bedrock-agentcore[strands-agents]'
```

```bash
git clone https://github.com/aws/bedrock-agentcore-sdk-python.git
cd bedrock-agentcore-sdk-python
uv sync
source .venv/bin/activate
```

--------------------------------

### Install Strands AgentCore SDK

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/integrations/strands/README.md

Install the necessary package for Strands AgentCore integration. Use the `[strands-agents]` extra for full functionality.

```bash
pip install 'bedrock-agentcore[strands-agents]'
```

--------------------------------

### LangGraph Agent Example

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/a2a_protocol_examples.md

Example of creating a LangGraph AgentExecutor and serving it with an AgentCard using `serve_a2a`.

```APIDOC
## LangGraph Agent Example

This example shows how to create a custom `AgentExecutor` for a LangGraph agent and serve it with an `AgentCard` using `serve_a2a`.

### Code
```python
from langchain_aws import ChatBedrockConverse
from langgraph.prebuilt import create_react_agent

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentSkill, Part, TextPart
from a2a.utils import new_task
from bedrock_agentcore.runtime import serve_a2a

llm = ChatBedrockConverse(model="us.anthropic.claude-sonnet-4-20250514")
graph = create_react_agent(llm, tools=[], prompt="You are a helpful calculator.")


class LangGraphA2AExecutor(AgentExecutor):
    def __init__(self, graph):
        self.graph = graph

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        task = context.current_task or new_task(context.message)
        if not context.current_task:
            await event_queue.enqueue_event(task)
        updater = TaskUpdater(event_queue, task.id, task.context_id)
        user_text = context.get_user_input()
        result = await self.graph.ainvoke({"messages": [("user", user_text)]})
        response = result["messages"][-1].content
        await updater.add_artifact([Part(root=TextPart(text=response))])
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        pass


card = AgentCard(
    name="langgraph-agent",
    description="A LangGraph agent on Bedrock AgentCore",
    url="http://localhost:9000/",
    version="0.1.0",
    capabilities=AgentCapabilities(streaming=True),
    skills=[AgentSkill(id="calc", name="calculator", description="Arithmetic", tags=["math"])],
    default_input_modes=["text"],
    default_output_modes=["text"],
)

if __name__ == "__main__":
    serve_a2a(LangGraphA2AExecutor(graph), card)
```
```

--------------------------------

### Existing AG-UI Boilerplate Code

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/proposals/ag-ui-protocol-support.md

This is the current example code for handling AG-UI protocol, which includes manual setup of FastAPI, EventEncoder, StreamingResponse, and uvicorn.

```python
app = FastAPI()

@app.post("/invocations")
async def invocations(input_data: dict, request: Request):
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        run_input = RunAgentInput(**input_data)
        async for event in agui_agent.run(run_input):
            yield encoder.encode(event)

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

@app.get("/ping")
async def ping():
    return JSONResponse({"status": "Healthy"})

uvicorn.run(app, host="0.0.0.0", port=8080)
```

--------------------------------

### Install Bedrock AgentCore SDK

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Install the SDK using pip. For Strands Agents integration, use the optional dependency.

```bash
pip install bedrock-agentcore
```

```bash
pip install 'bedrock-agentcore[strands-agents]'
```

--------------------------------

### Quick Start: Create Payment Instrument and Session

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Initialize PaymentManager and create a payment instrument (e.g., an embedded crypto wallet) and a payment session with defined spending limits.

```python
import os
from bedrock_agentcore.payments import PaymentManager

manager = PaymentManager(
    payment_manager_arn=os.environ["PAYMENT_MANAGER_ARN"],
    region_name="us-east-1",
)

# Create a payment instrument (embedded crypto wallet)
instrument = manager.create_payment_instrument(
    payment_connector_id=os.environ["PAYMENT_CONNECTOR_ID"],
    payment_instrument_type="EMBEDDED_CRYPTO_WALLET",
    payment_instrument_details={
        "embeddedCryptoWallet": {
            "network": "ETHEREUM",
            "linkedAccounts": [
                {"email": {"emailAddress": "user@example.com"}}
            ],
        }
    },
    user_id="user-123",
)

# Create a payment session with spending limits
session = manager.create_payment_session(
    expiry_time_in_minutes=60,
    user_id="user-123",
    limits={"maxSpendAmount": {"value": "100.00", "currency": "USD"}},
)
```

--------------------------------

### Strands Agent Example

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/a2a_protocol_examples.md

Example of how to wrap a Strands Agent as an A2A executor and serve it using `serve_a2a`.

```APIDOC
## Strands Agent Example

This example demonstrates how to use `StrandsA2AExecutor` to wrap a Strands `Agent` and serve it using the `serve_a2a` function from `bedrock_agentcore.runtime`.

### Code
```python
from strands import Agent
from strands.a2a import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a

agent = Agent(
    model="us.anthropic.claude-sonnet-4-20250514",
    system_prompt="You are a helpful calculator.",
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))
```
```

--------------------------------

### Install and Run Security Scanning Tools

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/SECURITY.md

Install Bandit and Safety for security scanning. Use Bandit to recursively scan the 'src/' directory and Safety to check for known vulnerabilities in dependencies.

```bash
# Install security scanning tools
pip install bandit safety
```

```bash
# Run security scan
bandit -r src/
```

```bash
# Check for known vulnerabilities
safety check
```

--------------------------------

### Install AgentCore with A2A Support - Bash

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

Install the bedrock-agentcore package with A2A protocol support using pip.

```bash
pip install "bedrock-agentcore[a2a]"
```

--------------------------------

### Install Strands AgentCore Evals

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Install the necessary package for Strands AgentCore evaluation.

```bash
pip install 'bedrock-agentcore[strands-agents-evals]'
```

--------------------------------

### Google ADK Agent Example

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/a2a_protocol_examples.md

Example of using Google ADK's `A2aAgentExecutor` with an explicit AgentCard and serving it via `serve_a2a`.

```APIDOC
## Google ADK Agent Example

This example demonstrates how to use Google ADK's `A2aAgentExecutor` with an `LlmAgent` and an explicit `AgentCard`, served using `serve_a2a`.

### Code
```python
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.a2a import A2aAgentExecutor

from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from bedrock_agentcore.runtime import serve_a2a

agent = LlmAgent(
    model="gemini-2.0-flash",
    name="calculator",
    description="A calculator agent",
    instruction="You are a helpful calculator.",
)
runner = Runner(agent=agent, app_name="calculator", session_service=None)

card = AgentCard(
    name="adk-agent",
    description="A Google ADK agent on Bedrock AgentCore",
    url="http://localhost:9000/",
    version="0.1.0",
    capabilities=AgentCapabilities(streaming=True),
    skills=[AgentSkill(id="calc", name="calculator", description="Arithmetic", tags=["math"])],
    default_input_modes=["text"],
    default_output_modes=["text"],
)

if __name__ == "__main__":
    serve_a2a(A2aAgentExecutor(runner=runner), card)
```
```

--------------------------------

### Create Payment Manager and Connector

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

This code snippet demonstrates the one-time setup for creating a payment manager and its associated connector. It uses the PaymentClient to provision resources and exports their ARNs for later use. The `wait_for_ready` parameter ensures all resources are ready before proceeding, with automatic rollback on failure.

```python
import os
from bedrock_agentcore.payments.client import PaymentClient

# This is typically done once, separately from your agent application
payment_client = PaymentClient(region_name="us-east-1")

response = payment_client.create_payment_manager_with_connector(
    payment_manager_name="AgentCorePaymentManager",
    payment_manager_description="Payment Manager for Agent Core",
    authorizer_type="AWS_IAM",
    role_arn="arn:aws:iam::123456789012:role/BedrockAgentCoreFullAccess",
    payment_connector_config={
        "name": "agent-core-connector",
        "description": "Payment Connector for Agent Core",
        "payment_credential_provider_config": {
            "name": "agent-core-provider",
            "credential_provider_vendor": "CoinbaseCDP",
            "credentials": {
                "api_key_id": "<your-coinbase-api-key-id>",
                "api_key_secret": "<your-coinbase-api-key-secret>",
                "wallet_secret": "<your-coinbase-wallet-secret>",
            },
        },
    },
    wait_for_ready=True,
    max_wait=300,
    poll_interval=5,
)

# Export for reuse in your agent application
payment_manager_arn = response["paymentManager"]["paymentManagerArn"]
payment_connector_id = response["paymentConnector"]["paymentConnectorId"]
os.environ["PAYMENT_MANAGER_ARN"] = payment_manager_arn
os.environ["PAYMENT_CONNECTOR_ID"] = payment_connector_id
print(f"Payment Manager ARN: {payment_manager_arn}")
print(f"Payment Connector ID: {payment_connector_id}")
```

--------------------------------

### serve_a2a Function Reference

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/a2a_protocol_examples.md

Reference for the `serve_a2a` function used to start a Bedrock-compatible A2A server.

```APIDOC
## `serve_a2a(executor, agent_card=None, *, port=9000, host=None, ...)`

Starts a Bedrock-compatible A2A server with `uvicorn`.

### Parameters

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `executor` | `AgentExecutor` | required | An a2a-sdk `AgentExecutor` that implements the agent logic |
| `agent_card` | `AgentCard` | `None` | Agent metadata. Auto-built from executor if omitted (works best with Strands) |
| `port` | `int` | `9000` | Port to serve on |
| `host` | `str` | `None` | Host to bind to. Auto-detected: `0.0.0.0` in Docker, `127.0.0.1` otherwise |
| `task_store` | `TaskStore` | `None` | Custom task store; defaults to `InMemoryTaskStore` |
| `context_builder` | `CallContextBuilder` | `None` | Custom context builder; defaults to `BedrockCallContextBuilder` |
| `ping_handler` | `Callable[[], PingStatus]` | `None` | Custom health check callback |
| `**kwargs` | | | Additional arguments forwarded to `uvicorn.run()` |
```

--------------------------------

### Connect to WebSocket with websockets library

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/agent_runtime_client_examples.md

Connect to the Agent Runtime WebSocket using the 'websockets' library. Ensure the library is installed and imported.

```python
import websockets

ws_url, headers = client.generate_ws_connection(runtime_arn)
async with websockets.connect(ws_url, extra_headers=headers) as ws:
    await ws.send(message)
```

--------------------------------

### Install Strands AgentCore Tool Search Plugin

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/gateway/integrations/strands/plugins/agentcore_tool_search/README.md

Install the plugin and its dependencies using pip. This command includes the necessary extras for Strands Agents integration.

```bash
pip install 'bedrock-agentcore[strands-agents]'
```

--------------------------------

### Dynamic Token Provider Setup

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

Demonstrates how to set up a dynamic token provider for automatic JWT refresh before each request, recommended for production environments. This involves defining a function to fetch a fresh token and passing it to the AgentCorePaymentsPluginConfig.

```APIDOC
## Dynamic Token Provider (Recommended for Production)

Use a callable token provider for automatic token refresh before each request:

```python
import requests

def get_fresh_token() -> str:
    """Fetch a fresh JWT from your identity provider."""
    resp = requests.post(
        "https://your-domain.auth.us-east-1.amazoncognito.com/oauth2/token",
        data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

config = AgentCorePaymentsPluginConfig(
    payment_manager_arn="arn:aws:bedrock-agentcore:us-east-1:123456789012:payment-manager/pm-jwt",
    token_provider=get_fresh_token,  # Called before each request
    payment_instrument_id="payment-instrument-xyz789",
    payment_session_id="payment-session-def456",
    region="us-east-1",
)

plugin = AgentCorePaymentsPlugin(config=config)
agent = Agent(
    system_prompt="You are a helpful assistant that can access paid APIs.",
    tools=[http_request],
    plugins=[plugin],
)

# 402 responses are handled automatically using JWT auth
agent("Fetch data from https://premium-api.example.com/data")
```

> **Note:** `bearer_token` and `token_provider` are mutually exclusive. Use `token_provider` in
> production for automatic token refresh. Use `bearer_token` for quick testing with a known token.
```

--------------------------------

### Initialize and Use AgentCoreToolSearchPlugin

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/gateway/integrations/strands/plugins/agentcore_tool_search/README.md

Demonstrates basic initialization of the AgentCoreToolSearchPlugin with an MCP client and its subsequent use within a Strands Agent. Ensure the MCP client is started before agent invocation.

```python
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
from strands import Agent
from strands.tools.mcp import MCPClient
from bedrock_agentcore.gateway.integrations.strands.plugins import AgentCoreToolSearchPlugin

mcp_client = MCPClient(lambda: aws_iam_streamablehttp_client(
    endpoint="https://<gateway-id>.gateway.bedrock-agentcore.<region>.amazonaws.com/mcp",
    aws_region="us-east-1",
    aws_service="bedrock-agentcore",
))

mcp_client.start()

agent = Agent(plugins=[AgentCoreToolSearchPlugin(mcp_client=mcp_client)])

agent("Find me afternoon flights to New York")
```

--------------------------------

### Setup Agent and In-Memory Telemetry

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Configure a Strands agent with tools and set up in-memory telemetry to capture OpenTelemetry spans for local evaluation.

```python
from strands import Agent, tool
from strands_evals import Experiment, Case
from strands_evals.telemetry import StrandsEvalsTelemetry
from bedrock_agentcore.evaluation import create_strands_evaluator

# Define your tools
@tool
def calculator(expression: str) -> str:
    """Evaluates a mathematical expression."""
    return str(eval(expression))

# Setup telemetry to capture spans
telemetry = StrandsEvalsTelemetry().setup_in_memory_exporter()

# Create your agent
agent = Agent(
    tools=[calculator],
    system_prompt="You are a helpful math assistant."
)
```

--------------------------------

### Use AgentCoreToolSearchPlugin with Context Manager

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/gateway/integrations/strands/plugins/agentcore_tool_search/README.md

Shows how to initialize and use the AgentCoreToolSearchPlugin within a Python context manager. This ensures proper setup and teardown of the MCP client.

```python
with mcp_client:
    agent = Agent(plugins=[AgentCoreToolSearchPlugin(mcp_client=mcp_client)])
    agent("Find me afternoon flights to New York")
```

--------------------------------

### List Events with Multiple Metadata and Branch Filters

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/metadata-workflow.ipynb

This example shows how to list events that match multiple metadata criteria (AND logic) within a specific branch, including parent branches. Ensure 'session', 'build_metadata_filter', and 'branch_name' are defined.

```python
# Example:
# Let us consider two metadata filters to be provided when listing events.
# key1 = "location", value1= "NYC"
# key2 = "Season", value1= "Winter"

metadata_filter_4 = build_metadata_filter(key="location", operator=OperatorType.EQUALS_TO, val="NYC")

metadata_filter_5 = build_metadata_filter(key="season", operator=OperatorType.EQUALS_TO, val="Winter")

filtered_events_5 = session.list_events(
    branch_name=branch_name, include_parent_branches=True, eventMetadata=[metadata_filter_4, metadata_filter_5]
)

print("=== Listing events with branch and metadata filters ===")
for index, event in enumerate(filtered_events_5, start=1):
    print(f"=== Event #{index} ===")
    print(event)
```

--------------------------------

### Setup In-Memory Exporter for Local Agents

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

For local agents, set up an in-memory telemetry exporter to capture spans. This is crucial for troubleshooting 'No trajectory data available' errors.

```python
telemetry = StrandsEvalsTelemetry().setup_in_memory_exporter()
# ... run agent ...
spans = telemetry.in_memory_exporter.get_finished_spans()
```

--------------------------------

### Manage Memory Records

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/README.md

Provides examples for listing, retrieving, and deleting memory records within a specified namespace. Ensure the namespace path is correctly formatted.

```python
# List all memory records in a namespace
records = session.list_long_term_memory_records(
    namespace_path="/user/preferences/user-123/",
    max_results=20
)

# Get specific memory record
record = session.get_memory_record("record-id-123")
print(f"Record content: {record.content}")

# Delete memory record
session.delete_memory_record("record-id-123")
```

--------------------------------

### BedrockAgentCore API Reference for Task Management

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/README.md

Provides Python code examples for key API methods related to manual task management, status control, and information retrieval within BedrockAgentCore.

```python
# Manual task management
task_id = app.add_async_task("task_name", metadata={"key": "value"})
success = app.complete_async_task(task_id)  # Returns True/False

# Status control
app.force_ping_status(PingStatus.HEALTHY)
app.clear_forced_ping_status()

# Information
status = app.get_current_ping_status()
info = app.get_async_task_info()
```

--------------------------------

### STM Example Conversation Flow

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/integrations/strands/README.md

Demonstrates a simple conversation where the agent remembers user preferences across turns. The agent's responses will be influenced by previously stated likes.

```python
agent("I like sushi with tuna")
# Agent remembers this preference

agent("I like pizza")
# Agent acknowledges both preferences

agent("What should I buy for lunch today?")
# Agent suggests options based on remembered preferences
```

--------------------------------

### Deploy Local Agent to Bedrock AgentCore

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

Use this snippet to deploy your local AI agent to Bedrock AgentCore. It requires the bedrock_agentcore library and an agent implementation (e.g., from 'strands'). The handler function processes incoming requests, streams agent responses, and the app.run() method starts the service.

```python
from bedrock_agentcore import BedrockAgentCoreApp
app = BedrockAgentCoreApp()

from strands import Agent # or bring your agent. 

@app.entrypoint
async def handler(request):
    prompt = request.get("prompt")

    agent = Agent()

    async for event in agent.stream_async(prompt):
        yield (event)

app.run()
```

--------------------------------

### Configure AG-UI App with Entrypoint and Run

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/proposals/ag-ui-protocol-support.md

Instantiate `AGUIApp` and explicitly set the agent's entrypoint, then run the application.

```python
from bedrock_agentcore.runtime import AGUIApp

app = AGUIApp()
app.entrypoint(agui_agent)
app.run()
```

--------------------------------

### Create Payment Manager and Connector

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Use this to set up a payment manager and connector. Ensure you have the necessary environment variables for credentials. The `wait_for_ready` parameter polls for resource readiness.

```python
import os
from bedrock_agentcore.payments.client import PaymentClient

payment_client = PaymentClient(region_name="us-east-1")

response = payment_client.create_payment_manager_with_connector(
    payment_manager_name="AgentCorePaymentManager",
    payment_manager_description="Payment Manager for Agent Core",
    authorizer_type="AWS_IAM",
    role_arn="arn:aws:iam::123456789012:role/BedrockAgentCoreFullAccess",
    payment_connector_config={
        "name": "agent-core-connector",
        "description": "Payment Connector for Agent Core",
        "payment_credential_provider_config": {
            "name": "agent-core-provider",
            "credential_provider_vendor": "CoinbaseCDP",
            "credentials": {
                "api_key_id": os.environ["COINBASE_API_KEY_ID"],
                "api_key_secret": os.environ["COINBASE_API_KEY_SECRET"],
                "wallet_secret": os.environ["COINBASE_WALLET_SECRET"],
            },
        },
    },
    wait_for_ready=True,
    max_wait=300,
    poll_interval=5,
)

# Extract details from response
payment_manager_arn = response["paymentManager"]["paymentManagerArn"]
payment_connector_id = response["paymentConnector"]["paymentConnectorId"]
credential_provider_arn = response["credentialProvider"]["credentialProviderArn"]
print(f"Payment Manager ARN: {payment_manager_arn}")
print(f"Payment Connector ID: {payment_connector_id}")
print(f"Credential Provider ARN: {credential_provider_arn}")
```

--------------------------------

### Create and Deploy Agent with CLI - Bash

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

Use the AgentCore CLI to create a new agent with default settings, navigate to its directory, and deploy it.

```bash
agentcore create --name MyAgent --defaults
cd MyAgent
agentcore deploy
```

--------------------------------

### Initialize Memory and Session

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/metadata-workflow.ipynb

Sets up the region, memory manager, and memory session for storing conversational data.

```python
region = "us-west-2"
```

```python
memory_manager = MemoryManager(region_name=region)

memory = memory_manager.get_or_create_memory(name="travel_support_agent_1")
```

```python
session_manager = MemorySessionManager(memory_id=memory["id"], region_name=region)

session = session_manager.create_memory_session(actor_id="user-123", session_id="session-1")
```

--------------------------------

### Initialize AgentCoreToolSearchPlugin with Custom System Prompt

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/gateway/integrations/strands/plugins/agentcore_tool_search/README.md

Shows how to configure the AgentCoreToolSearchPlugin with a custom system prompt for the StrandsIntentProvider, influencing how user intent is classified.

```python
from bedrock_agentcore.gateway.integrations.strands.plugins import AgentCoreToolSearchPlugin
from bedrock_agentcore.gateway.integrations.strands.plugins.agentcore_tool_search.intent_providers import StrandsIntentProvider

agent = Agent(plugins=[
    AgentCoreToolSearchPlugin(
        mcp_client=mcp_client,
        intent_provider=StrandsIntentProvider(
            system_prompt="Classify the user's intent in one sentence. Focus on the action, not details."
        ),
    )
])
```

--------------------------------

### Configure and Use AgentCore Payments Plugin

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

This Python code demonstrates setting up and using the AgentCorePaymentsPlugin to interact with tools via the AgentCore Gateway. It includes obtaining an OAuth token, initializing the MCP client, configuring the payment plugin with necessary details, and creating an agent that automatically handles 402 payments.

```python
import os
from dotenv import load_dotenv
load_dotenv()

from datetime import timedelta
import requests as http_requests
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from bedrock_agentcore.payments.integrations.strands import (
    AgentCorePaymentsPlugin,
    AgentCorePaymentsPluginConfig,
)

GATEWAY_URL           = os.environ["GATEWAY_URL"]
CLIENT_ID             = os.environ["CLIENT_ID"]
CLIENT_SECRET         = os.environ["CLIENT_SECRET"]
TOKEN_URL             = os.environ["TOKEN_URL"]
MODEL_ID              = os.environ.get("MODEL_ID", "us.anthropic.claude-sonnet-4-20250514-v1:0")
PAYMENT_MANAGER_ARN   = os.environ["PAYMENT_MANAGER_ARN"]
USER_ID               = os.environ["USER_ID"]
PAYMENT_INSTRUMENT_ID = os.environ["PAYMENT_INSTRUMENT_ID"]
PAYMENT_SESSION_ID    = os.environ["PAYMENT_SESSION_ID"]
REGION                = os.environ.get("AWS_REGION", "us-west-2")

def get_oauth_token():
    resp = http_requests.post(TOKEN_URL, data={
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }, headers={"Content-Type": "application/x-www-form-urlencoded"})
    resp.raise_for_status()
    return resp.json()["access_token"]

def main():
    token = get_oauth_token()

    # 1. Connect to Gateway MCP server
    mcp_client = MCPClient(lambda: streamablehttp_client(
        GATEWAY_URL,
        headers={"Authorization": f"Bearer {token}"},
        timeout=timedelta(seconds=120),
    ))

    # 2. Configure payment plugin
    payment_plugin = AgentCorePaymentsPlugin(config=AgentCorePaymentsPluginConfig(
        payment_manager_arn=PAYMENT_MANAGER_ARN,
        user_id=USER_ID,
        payment_instrument_id=PAYMENT_INSTRUMENT_ID,
        payment_session_id=PAYMENT_SESSION_ID,
        region=REGION,
    ))

    # 3. Create agent — plugin handles 402 payments automatically
    with mcp_client:
        agent = Agent(
            model=BedrockModel(model_id=MODEL_ID, streaming=True),
            tools=mcp_client.list_tools_sync(),
            plugins=[payment_plugin],
        )
        result = agent("Get me the latest crypto news")
        print(result.message)

if __name__ == "__main__":
    main()

```

--------------------------------

### Serve AG-UI Agent with Default Configuration

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/proposals/ag-ui-protocol-support.md

Use `serve_ag_ui` to quickly expose an existing agent with a `.run()` method for AG-UI protocol support.

```python
from bedrock_agentcore.runtime import serve_ag_ui

serve_ag_ui(agui_agent)
```

--------------------------------

### Import Core Components

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/integrations/strands/README.md

Import necessary classes and modules for setting up memory clients, agents, and configurations.

```python
import uuid
import boto3
from datetime import date
from strands import Agent
from bedrock_agentcore.memory import MemoryClient
from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager
```

--------------------------------

### Stripe Privy Credentials Configuration

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Example configuration for Stripe Privy credentials. Use this when `credential_provider_vendor` is set to `StripePrivy`.

```python
"credentials": {
    "app_id": os.environ["STRIPE_PRIVY_APP_ID"],
    "app_secret": os.environ["STRIPE_PRIVY_APP_SECRET"],
    "authorization_key": os.environ.get("STRIPE_PRIVY_AUTH_KEY", ""),  # optional
}
```

--------------------------------

### Initialize and Manage Memory Sessions

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/README.md

Demonstrates initializing the MemorySessionManager, creating a new memory session for an actor, and adding conversation turns. Use this to set up your memory management and log initial interactions.

```python
from bedrock_agentcore.memory import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole

# Initialize the session manager
manager = MemorySessionManager(
    memory_id="your-memory-id",  # Use existing memory id
    region_name="us-east-1"
)

# Create a session for a specific actor
session = manager.create_memory_session(
    actor_id="user-123",
    session_id="session-456"  # Optional - will generate UUID if not provided
)

# Add conversation turns
session.add_turns([
    ConversationalMessage("I love eating apples and cherries", MessageRole.USER),
    ConversationalMessage("Apples are very good for you!", MessageRole.ASSISTANT),
    ConversationalMessage("What's your favorite thing about apples?", MessageRole.USER),
    ConversationalMessage("I enjoy their flavor and nutritional benefits", MessageRole.ASSISTANT)
])

# Search long-term memories (after memory extraction has occurred)
memories = session.search_long_term_memories(
    query="what food does the user like",
    namespace_path="/food/user-123/",
    top_k=5
)

# Or search across multiple users
memories = manager.search_long_term_memories(
    query="Food preferences",
    namespace_path="/food/",  # Search all food-related memories
    top_k=10
)
```

--------------------------------

### Handle Authentication Errors

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/README.md

Demonstrates how to catch and handle `NoCredentialsError` when initializing the MemorySessionManager, guiding the user to configure AWS credentials.

```python
try:
    manager = MemorySessionManager(memory_id="test")
except NoCredentialsError:
    # Guide user to configure credentials
    print("Please run 'aws configure' or set AWS environment variables")
```

--------------------------------

### Create Strands Evaluator API Reference

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Example demonstrating the creation of a Strands evaluator with specified ID, region, and test pass score.

```python
evaluator = create_strands_evaluator(
    "Builtin.Helpfulness",
    region="us-east-1",
    test_pass_score=0.8
)
```

--------------------------------

### Serve AG-UI Protocol with Bedrock AgentCore

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/README.md

This snippet demonstrates how to serve agents using the AG-UI protocol over both SSE and WebSocket. It utilizes the `serve_ag_ui` function from the bedrock_agentcore.runtime module.

```python
from bedrock_agentcore.runtime import serve_ag_ui

```

--------------------------------

### Get Payment Session Details

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Check the status and remaining payment limits of an existing payment session using its ID and the user ID.

```python
session_details = manager.get_payment_session(
    payment_session_id=payment_session_id,
    user_id="test-user-123",
)
print(f"Available: {session_details.get('availableLimits', {}).get('availableSpendAmount')}")
```

--------------------------------

### Get Payment Session Details

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Retrieves the current status and remaining limits of a specific payment session using its ID and the associated user ID.

```APIDOC
## Get Payment Session Details

### Description
Retrieves details of a specific payment session, including available payment limits.

### Method
`manager.get_payment_session`

### Parameters
- **payment_session_id** (str) - Required - The ID of the payment session to retrieve.
- **user_id** (str) - Required - The identifier for the user associated with the payment session.

### Response Example
```json
{
  "availableLimits": {
    "availableSpendAmount": "95.50"
  }
}
```
```

--------------------------------

### Initialize Payments Plugin with Dynamic Selection

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

This snippet demonstrates initializing the AgentCorePaymentsPlugin without a predefined payment instrument or session ID, allowing for dynamic resolution during runtime. The plugin is then integrated into an Agent.

```python
config = AgentCorePaymentsPluginConfig(
    payment_manager_arn="arn:aws:bedrock-agentcore:us-east-1:123456789012:payment-manager/pm-abc123",
    user_id="user-123",
    region="us-east-1",
    # payment_instrument_id and payment_session_id omitted
)

plugin = AgentCorePaymentsPlugin(config=config)
agent = Agent(
    system_prompt="You are a helpful assistant.",
    tools=[http_request],
    plugins=[plugin],
)
```

--------------------------------

### Test Business Actions with Curl

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/TESTING_GUIDE.md

Trigger business logic actions such as starting a background task or retrieving task information using curl and the /invocations endpoint.

```bash
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"action": "start_background_task"}'
```

```bash
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"action": "get_task_info"}'
```

--------------------------------

### Control Plane Operations

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

Provides examples of common control plane operations for managing payment resources, including creating, listing, and updating payment managers and connectors.

```APIDOC
## Control Plane Operations

For individual resource management (alternative to `create_payment_manager_with_connector`):

```python
from bedrock_agentcore.payments.client import PaymentClient

client = PaymentClient(region_name="us-east-1")

# Create a payment manager
manager_response = client.create_payment_manager(
    name="MyPaymentManager",
    role_arn="arn:aws:iam::123456789012:role/PaymentRole",
    authorizer_type="AWS_IAM",
    wait_for_ready=True,
)

# Create a payment connector
connector_response = client.create_payment_connector(
    payment_manager_id=manager_response["paymentManagerId"],
    name="MyCoinbaseConnector",
    connector_type="CoinbaseCDP",
    credential_provider_configurations=[
        {"coinbaseCDP": {"credentialProviderArn": "arn:..."}}
    ],
    wait_for_ready=True,
)

# List payment managers
managers = client.list_payment_managers(max_results=10)

# List connectors for a manager
connectors = client.list_payment_connectors(
    payment_manager_id=manager_response["paymentManagerId"],
)

# Update a payment manager
client.update_payment_manager(
    payment_manager_id=manager_response["paymentManagerId"],
    description="Updated description",
)
```
```

--------------------------------

### Configure and Initialize Agent for STM

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/integrations/strands/README.md

Set up the AgentCoreMemoryConfig with memory, session, and actor IDs. Then, initialize the AgentCoreMemorySessionManager and the Strands Agent.

```python
MEM_ID = basic_memory.get('id')
ACTOR_ID = "actor_id_test_%s" % datetime.now().strftime("%Y%m%d%H%M%S")
SESSION_ID = "testing_session_id_%s" % datetime.now().strftime("%Y%m%d%H%M%S")


# Configure memory
agentcore_memory_config = AgentCoreMemoryConfig(
    memory_id=MEM_ID,
    session_id=SESSION_ID,
    actor_id=ACTOR_ID
)

# Create session manager
session_manager = AgentCoreMemorySessionManager(
    agentcore_memory_config=agentcore_memory_config,
    region_name="us-east-1"
)

# Create agent
agent = Agent(
    system_prompt="You are a helpful assistant. Use all you know about the user to provide helpful responses.",
    session_manager=session_manager,
)
```

--------------------------------

### Manual Task Management

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/README.md

Manually manage the tracking of asynchronous tasks using `app.add_async_task` to start tracking and `app.complete_async_task` to stop tracking. This provides explicit control over task lifecycle.

```APIDOC
## Manual Task Management

Manually control task tracking:

```python
@app.entrypoint
async def handler(event):
    # Start tracking
    task_id = app.add_async_task("data_processing", {"batch": 100})

    # Do work
    process_data()

    # Stop tracking
    app.complete_async_task(task_id)
    return {"status": "completed"}
```
```

--------------------------------

### Daily Development Workflow with Git Hooks

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/CONTRIBUTING.md

Illustrates the daily development workflow, including making changes, committing with pre-commit hooks, and pushing code. Hooks automatically run formatting, linting, security scanning, and tests.

```bash
# Make your changes
vim src/bedrock_agentcore/myfile.py

# Commit (hooks run automatically)
git commit -m "feat: add new feature"
# ↑ Formatting and linting run here

# Push (tests run automatically)
git push origin my-branch
# ↑ Security scanning and tests run here
```

--------------------------------

### Manual Task Management with BedrockAgentCore

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/README.md

Manually control the tracking of asynchronous tasks using `app.add_async_task` to start tracking and `app.complete_async_task` to stop. This provides explicit control over task lifecycle and reporting.

```python
from bedrock_agent_core.app import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

def process_data():
    # Replace with actual data processing logic
    pass

@app.entrypoint
async def handler(event):
    # Start tracking
    task_id = app.add_async_task("data_processing", {"batch": 100})

    # Do work
    process_data()

    # Stop tracking
    app.complete_async_task(task_id)
    return {"status": "completed"}
```

--------------------------------

### Basic AgentCoreToolSearchPlugin Initialization

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/gateway/integrations/strands/plugins/agentcore_tool_search/README.md

Illustrates the minimal configuration required to initialize the AgentCoreToolSearchPlugin. It automatically uses the parent agent's model for intent classification.

```python
from bedrock_agentcore.gateway.integrations.strands.plugins import AgentCoreToolSearchPlugin

agent = Agent(plugins=[
    AgentCoreToolSearchPlugin(mcp_client=mcp_client)
])
```

--------------------------------

### Initialize Strands Agent with Payments Plugin

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/integrations/strands/README.md

Configure and initialize the AgentCorePaymentsPlugin with payment manager details and user information. Then, integrate it into a Strands Agent instance.

```python
import os
from strands import Agent
from strands_tools import http_request
from bedrock_agentcore.payments.integrations.strands import (
    AgentCorePaymentsPlugin,
    AgentCorePaymentsPluginConfig,
)

plugin = AgentCorePaymentsPlugin(config=AgentCorePaymentsPluginConfig(
    payment_manager_arn=os.environ["PAYMENT_MANAGER_ARN"],
    user_id="test-user-123",
    payment_instrument_id=os.environ["PAYMENT_INSTRUMENT_ID"],
    payment_session_id=os.environ["PAYMENT_SESSION_ID"],
    region="us-east-1",
))

agent = Agent(
    system_prompt="You are a helpful assistant that can access paid APIs.",
    tools=[http_request],
    plugins=[plugin],
)

# 402 responses are automatically handled
agent("Fetch a joke from https://premium-api.example.com/joke")
```

--------------------------------

### Manage Conversation Branches and History

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/memory/README.md

Provides examples for retrieving conversation history, forking conversations into new branches, listing all available branches, and retrieving events from a specific branch. Use this for complex dialogue flows and scenario management.

```python
# Get conversation history
turns = session.get_last_k_turns(k=3)
print(f"Last 3 conversation turns: {len(turns)}")

# Fork conversation for alternative scenario
branch_event = session.fork_conversation(
    root_event_id="event-123",
    branch_name="premium-option",
    messages=[
        ConversationalMessage("What about expedited shipping?", MessageRole.USER),
        ConversationalMessage("I can upgrade you to overnight delivery for $20", MessageRole.ASSISTANT)
    ]
)

# List all branches in the session
branches = session.list_branches()
for branch in branches:
    print(f"Branch: {branch.name}, Events: {branch.event_count}")

# Get events from specific branch
branch_events = session.list_events(branch_name="premium-option")
```

--------------------------------

### Custom Boto3 Session for Client Initialization

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/docs/examples/agent_runtime_client_examples.md

Initializes the `AgentCoreRuntimeClient` with a custom `boto3.Session` object. This allows for custom credential handling, such as using specific profiles or explicit credentials.

```python
import boto3
from bedrock_agentcore.runtime import AgentCoreRuntimeClient

# Create a custom session with specific profile
session = boto3.Session(profile_name="my-profile")

# Or with specific credentials
session = boto3.Session(
    aws_access_key_id="YOUR_ACCESS_KEY",
    aws_secret_access_key="YOUR_SECRET_KEY",
    aws_session_token="YOUR_SESSION_TOKEN"
)

# Initialize client with custom session
client = AgentCoreRuntimeClient(region="us-west-2", session=session)

# Use the client normally
ws_url, headers = client.generate_ws_connection(runtime_arn)
```

--------------------------------

### GitHub Actions Workflow for Integration Tests

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/payments/README.md

A GitHub Actions workflow to automatically run integration tests on push and pull request events. It checks out code, sets up Python, installs dependencies, and runs tests with necessary environment variables configured.

```yaml
name: Integration Tests

on: [push, pull_request]

jobs:
  integration-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - uses: actions/setup-python@v2
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run integration tests
        env:
          BEDROCK_AGENTCORE_CONTROL_ENDPOINT: ${{ secrets.BEDROCK_ENDPOINT }}
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
        run: pytest tests_integ/payment/ -v
```

--------------------------------

### Run Comprehensive Async Status Tests

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/tests_integ/async/TESTING_GUIDE.md

From the async directory, execute the test script to validate all async functionality. Use the --quick flag for a shorter validation test.

```bash
python test_async_status_example.py --quick
```

```bash
python test_async_status_example.py
```

--------------------------------

### Create Payment Manager and Connector

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/payments/README.md

This snippet demonstrates how to create a payment manager and a payment connector. It includes options for configuring credentials for Coinbase CDP and Stripe Privy. The `wait_for_ready` parameter ensures resources are ready before proceeding.

```APIDOC
## Create Payment Manager and Connector

### Description
Creates a payment manager and a payment connector. This is typically a one-time setup.

### Method
`PaymentClient.create_payment_manager_with_connector`

### Parameters
- `payment_manager_name` (string) - Required - Name for the payment manager.
- `payment_manager_description` (string) - Optional - Description for the payment manager.
- `authorizer_type` (string) - Required - Type of authorizer (e.g., "AWS_IAM").
- `role_arn` (string) - Required - ARN of the IAM role.
- `payment_connector_config` (object) - Required - Configuration for the payment connector.
  - `name` (string) - Required - Name for the connector.
  - `description` (string) - Optional - Description for the connector.
  - `payment_credential_provider_config` (object) - Required - Configuration for the credential provider.
    - `name` (string) - Required - Name for the credential provider.
    - `credential_provider_vendor` (string) - Required - Vendor of the credential provider (e.g., "CoinbaseCDP", "StripePrivy").
    - `credentials` (object) - Required - Credentials for the provider.
      - For CoinbaseCDP: `api_key_id`, `api_key_secret`, `wallet_secret`.
      - For StripePrivy: `app_id`, `app_secret`, `authorization_key` (optional).
- `wait_for_ready` (boolean) - Optional - If true, polls until resources are ready.
- `max_wait` (integer) - Optional - Maximum time to wait for resources to be ready.
- `poll_interval` (integer) - Optional - Interval in seconds for polling.

### Response
Returns a dictionary containing details of the created payment manager, payment connector, and credential provider.

### Request Example
```python
import os
from bedrock_agentcore.payments.client import PaymentClient

payment_client = PaymentClient(region_name="us-east-1")

response = payment_client.create_payment_manager_with_connector(
    payment_manager_name="AgentCorePaymentManager",
    payment_manager_description="Payment Manager for Agent Core",
    authorizer_type="AWS_IAM",
    role_arn="arn:aws:iam::123456789012:role/BedrockAgentCoreFullAccess",
    payment_connector_config={
        "name": "agent-core-connector",
        "description": "Payment Connector for Agent Core",
        "payment_credential_provider_config": {
            "name": "agent-core-provider",
            "credential_provider_vendor": "CoinbaseCDP",
            "credentials": {
                "api_key_id": os.environ["COINBASE_API_KEY_ID"],
                "api_key_secret": os.environ["COINBASE_API_KEY_SECRET"],
                "wallet_secret": os.environ["COINBASE_WALLET_SECRET"],
            },
        },
    },
    wait_for_ready=True,
    max_wait=300,
    poll_interval=5,
)

payment_manager_arn = response["paymentManager"]["paymentManagerArn"]
payment_connector_id = response["paymentConnector"]["paymentConnectorId"]
credential_provider_arn = response["credentialProvider"]["credentialProviderArn"]
print(f"Payment Manager ARN: {payment_manager_arn}")
print(f"Payment Connector ID: {payment_connector_id}")
print(f"Credential Provider ARN: {credential_provider_arn}")
```
```

--------------------------------

### Fetch Spans from CloudWatch for Custom Agent

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Fetches ADOT spans and log records from CloudWatch for a specific agent session. Requires session ID, CloudWatch log group for event logs, and a start time. Use this for custom agent log groups.

```python
spans = fetch_spans_from_cloudwatch(
    session_id="abc-123",
    event_log_group="/my-app/agent-events",
    start_time=start_time
)
```

--------------------------------

### Fetch Spans from CloudWatch for Runtime Agent

Source: https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/src/bedrock_agentcore/evaluation/integrations/strands_agents_evals/README.md

Fetches ADOT spans and log records from CloudWatch for a specific agent session. Requires session ID, CloudWatch log group for event logs, and a start time. The log group format for runtime agents is specific.

```python
from bedrock_agentcore.evaluation import fetch_spans_from_cloudwatch
from datetime import datetime, timedelta, timezone

start_time = datetime.now(timezone.utc) - timedelta(minutes=10)
spans = fetch_spans_from_cloudwatch(
    session_id="abc-123",
    event_log_group="/aws/bedrock-agentcore/runtimes/my-agent-ABC123-DEFAULT",
    start_time=start_time
)
```