strands-agentcore-tools 0.1.5

Strands tools for AWS Bedrock AgentCore lifecycle management

×

Strands AgentCore Tools

Agentic tools that enable AI agents to autonomously deploy, manage, and monitor themselves on AWS Bedrock AgentCore

---

What This Is

9 Python functions that wrap AWS Bedrock AgentCore boto3 APIs for deployment, invocation, monitoring, and lifecycle management. Use these tools to:

• ✅ Deploy agents to AgentCore from code
• ✅ Invoke deployed agents programmatically
• ✅ Monitor CloudWatch logs and metrics
• ✅ Manage memory, sessions, and OAuth
• ✅ Integrate with Strands agents for autonomous deployment

Lightweight by design: Import only the tools you need for your use case.

---

Agent Example

Copy-paste ready agent with all execution modes:

# agent.py
import threading
from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke(payload, context):
    """Async entrypoint with yield support for streaming"""
    agent = Agent()
    prompt = payload.get("prompt", "")
    mode = payload.get("mode", "streaming")  # streaming | sync | fire_and_forget
    
    if mode == "fire_and_forget":
        # Start background thread
        task_id = app.add_async_task("agent_processing", payload)
        thread = threading.Thread(
            target=run_agent_background,
            args=(agent, prompt, task_id),
            daemon=True
        )
        thread.start()
        yield {"status": "started", "content": [{"text": "Agent running in background"}]}
    
    elif mode == "sync":
        # Single blocking response (wait for completion)
        result = agent(prompt)
        yield {"status": "success", "content": [{"text": str(result)}]}
    
    else:
        # Streaming response
        for event in agent.stream_async(prompt):
            yield event

def run_agent_background(agent, prompt, task_id):
    """Background worker (sync function, runs in separate thread)"""
    try:
        for event in agent.stream_async(prompt):
            print(event)  # Log events in background
        app.complete_async_task(task_id)
    except Exception as e:
        app.logger.error(f"Background task failed: {e}")
        app.complete_async_task(task_id)

app.run()

Deploy this agent:

from strands_agentcore_tools import configure, launch, invoke

configure(action="configure", entrypoint="agent.py", agent_name="my-agent")
launch(action="launch", agent_name="my-agent")
invoke(agent_arn="arn:...", payload='{"prompt": "Hello!", "mode": "streaming"}')

📝 Advanced: Agent with Memory (STM + LTM)

# agent.py - Agent with persistent memory
import os
from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent
from bedrock_agentcore.memory.integrations.strands.config import (
    AgentCoreMemoryConfig, RetrievalConfig
)
from bedrock_agentcore.memory.integrations.strands.session_manager import (
    AgentCoreMemorySessionManager
)
from strands_tools.agent_core_memory import AgentCoreMemoryToolProvider

app = BedrockAgentCoreApp()

MEMORY_ID = os.getenv("BEDROCK_AGENTCORE_MEMORY_ID")
REGION = os.getenv("AWS_REGION", "us-west-2")

@app.entrypoint
async def invoke(payload, context):
    """Async entrypoint with yield support"""
    session_id = context.session_id
    actor_id = context.headers.get(
        "X-Amzn-Bedrock-AgentCore-Runtime-Custom-Actor-Id", 
        "user"
    )
    
    # Configure memory with semantic search
    memory_config = AgentCoreMemoryConfig(
        memory_id=MEMORY_ID,
        session_id=session_id,
        actor_id=actor_id,
        retrieval_config={
            f"/users/{actor_id}/facts": RetrievalConfig(
                top_k=3, 
                relevance_score=0.5
            )
        }
    )
    
    # Memory tools provider
    memory_provider = AgentCoreMemoryToolProvider(
        memory_id=MEMORY_ID,
        session_id=session_id,
        actor_id=actor_id,
        namespace="default",
        region=REGION
    )
    
    # Create agent with memory
    agent = Agent(
        tools=memory_provider.tools,
        session_manager=AgentCoreMemorySessionManager(
            agentcore_memory_config=memory_config,
            region=REGION
        ),
        system_prompt="You have persistent memory across conversations."
    )
    
    # Stream responses
    for event in agent.stream_async(payload.get("prompt")):
        yield event

app.run()

Deploy with memory:

from strands_agentcore_tools import memory, configure, launch

# 1. Create memory
memory(
    action="create",
    name="my-memory",
    strategies=[{
        "semanticMemoryStrategy": {
            "name": "Facts",
            "namespaces": ["/users/{actorId}/facts"]
        }
    }],
    wait_for_active=True
)

# 2. Configure with memory
configure(
    action="configure",
    entrypoint="agent.py",
    agent_name="my-agent",
    memory_mode="STM_AND_LTM"
)

# 3. Launch
launch(action="launch", agent_name="my-agent")

---

Installation

pip install strands-agentcore-tools

Requirements:

• Python 3.10+
• AWS credentials configured
• IAM permissions for bedrock-agentcore:*, ecr:*, codebuild:*, iam:*, logs:*

---

The 9 Tools

Tool	Purpose	Key Parameters
configure	Generate Dockerfile, IAM roles, config YAML	entrypoint, agent_name, memory_mode
launch	Build & deploy to AgentCore via CodeBuild	agent_name, mode, env_vars
invoke	Execute deployed agent with payload	agent_arn, payload, session_id
agents	List/get agent runtimes	action ("list", "get", "find_by_name")
status	Check agent health & endpoint status	agent_id or agent_name
logs	Query CloudWatch logs	agent_name, action, filter_pattern
memory	Manage AgentCore memories (STM/LTM)	action, memory_id, strategies
identity	OAuth provider management	action, provider_type, vendor
session	Stop active runtime sessions	agent_arn, session_id

---

Quick Reference

1. Configure

Generates deployment files: Dockerfile, .bedrock_agentcore.yaml, IAM roles

from strands_agentcore_tools import configure

configure(
    action="configure",
    entrypoint="agent.py",
    agent_name="my-agent",
    memory_mode="STM_AND_LTM",          # or "STM_ONLY", "NO_MEMORY"
    enable_observability=True,
    idle_timeout=1800,                  # seconds
    max_lifetime=7200,                  # seconds
    region="us-west-2"
)

Actions:

• configure - Generate deployment files
• status - Check configuration status
• list - List all configured agents

Output:

• .bedrock_agentcore/<agent_name>/Dockerfile
• .bedrock_agentcore/<agent_name>/.dockerignore
• .bedrock_agentcore.yaml

---

2. Launch

Builds ARM64 container on CodeBuild, pushes to ECR, deploys to AgentCore

from strands_agentcore_tools import launch

result = launch(
    action="launch",
    agent_name="my-agent",
    mode="codebuild",                   # or "local" (requires Docker/Finch/Podman)
    auto_update_on_conflict=True,
    env_vars={
        "MODEL_ID": "us.anthropic.claude-sonnet-4-5-20250929-v1:0",
        "LOG_LEVEL": "DEBUG"
    },
    region="us-west-2"
)

print(result["agent_arn"])  # arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc

Actions:

• launch - Deploy agent
• status - Check deployment status
• stop - Stop active deployment

Returns:

{
    "agent_id": "my-agent-abc123",
    "agent_arn": "arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc123",
    "status": "READY"
}

---

3. Invoke

Execute deployed agent, supports streaming responses

from strands_agentcore_tools import invoke

# Streaming invocation (default)
invoke(
    agent_arn="arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc",
    payload='{"prompt": "Analyze this dataset"}',
    session_id="session-123",           # Optional: for conversation continuity
    user_id="user-abc",                 # Optional: for OAuth flows
    headers={                           # Optional: custom headers
        "X-Custom-Header": "value"
    },
    region="us-west-2"
)

Parameters:

• agent_arn (required) - Full ARN of deployed agent
• payload (required) - JSON string payload
• session_id (optional) - Min 33 chars for session tracking
• user_id (optional) - Triggers OAuth if configured
• headers (optional) - Custom HTTP headers
• region (optional) - AWS region

Returns: SSE streaming events from agent

---

4. Agents

List, get, and search agent runtimes

from strands_agentcore_tools import agents

# List all agents
agents(action="list", region="us-west-2")

# Get specific agent
agents(action="get", agent_id="my-agent-abc123", region="us-west-2")

# Find by name
result = agents(action="find_by_name", agent_name="my-agent", region="us-west-2")
print(result["agent_arn"])

Actions:

• list - List all agent runtimes
• get - Get agent by ID
• find_by_name - Find agent by name (returns first match)

Returns:

{
    "agent_id": "my-agent-abc123",
    "agent_arn": "arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc123",
    "status": "READY",
    "created": "2024-10-24T14:20:00Z"
}

---

5. Status

Check agent health and endpoint status

from strands_agentcore_tools import status

status(
    agent_id="my-agent-abc123",         # Can use agent_id
    agent_name="my-agent",              # Or agent_name (auto-lookup)
    endpoint="DEFAULT",                 # Optional: specific endpoint
    region="us-west-2"
)

Returns:

{
    "agent_id": "my-agent-abc123",
    "status": "READY",                  # READY, CREATING, UPDATING, DELETING, FAILED
    "endpoint": "DEFAULT",
    "created": "2024-10-24T14:20:00Z",
    "updated": "2024-10-24T15:30:00Z"
}

---

6. Logs

Query CloudWatch logs for deployed agents

from strands_agentcore_tools import logs

# Recent logs
logs(
    agent_name="my-agent",
    action="recent",
    limit=50,
    region="us-west-2"
)

# Search for errors
logs(
    agent_name="my-agent",
    action="search",
    filter_pattern="ERROR",
    limit=100,
    region="us-west-2"
)

# Tail logs (real-time)
logs(
    agent_name="my-agent",
    action="tail",
    region="us-west-2"
)

# List log streams
logs(
    agent_name="my-agent",
    action="streams",
    region="us-west-2"
)

Actions:

• recent - Get recent log events
• search - Filter logs by pattern
• tail - Stream logs in real-time
• streams - List available log streams

Parameters:

• agent_name - Agent name (auto-lookups runtime ID)
• agent_id - Or use runtime ID directly
• endpoint - Default: "DEFAULT"
• limit - Max events to return
• filter_pattern - CloudWatch filter pattern

---

7. Memory

Manage AgentCore memories (STM/LTM)

from strands_agentcore_tools import memory

# Create memory
memory(
    action="create",
    name="research-memory",
    strategies=[{
        "semanticMemoryStrategy": {
            "name": "Facts",
            "namespaces": ["/users/{actorId}/facts"]
        }
    }],
    wait_for_active=True,               # Block until ACTIVE
    region="us-west-2"
)

# Get memory status
memory(
    action="get_status",
    memory_id="memory-abc123",
    region="us-west-2"
)

# Retrieve from memory
memory(
    action="retrieve",
    memory_id="memory-abc123",
    namespace="/users/user-abc/facts",
    search_query="What is the capital of France?",
    top_k=5,
    min_score=0.5,
    region="us-west-2"
)

# Create memory event
memory(
    action="create_event",
    memory_id="memory-abc123",
    session_id="session-123",
    actor_id="user-abc",
    namespace="/users/user-abc/facts",
    content="Paris is the capital of France",
    region="us-west-2"
)

# List all memories
memory(action="list", region="us-west-2")

Actions:

• create - Create new memory
• get_status - Check memory status
• retrieve - Semantic search
• create_event - Add memory event
• list - List all memories

---

8. Identity

OAuth provider management

from strands_agentcore_tools import identity

# Create OAuth provider
identity(
    action="create",
    provider_type="oauth2",
    name="slack-oauth",
    vendor="SlackOauth2",               # SlackOauth2, GitHubOAuth2, GoogleOAuth2
    client_id="your-client-id",
    client_secret="your-client-secret",
    region="us-west-2"
)

# Get provider
identity(
    action="get",
    provider_id="provider-abc123",
    region="us-west-2"
)

# List providers
identity(action="list", region="us-west-2")

# Delete provider
identity(
    action="delete",
    provider_id="provider-abc123",
    region="us-west-2"
)

Actions:

• create - Create OAuth provider
• get - Get provider details
• list - List all providers
• delete - Delete provider

Supported Vendors:

• SlackOauth2
• GitHubOAuth2
• GoogleOAuth2

---

9. Session

Stop active runtime sessions

from strands_agentcore_tools import session

session(
    action="stop",
    agent_arn="arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc",
    session_id="session-123",
    region="us-west-2"
)

Actions:

• stop - Terminate active session

---

Complete Workflow Example

from strands_agentcore_tools import (
    configure, launch, invoke, status, logs, memory
)

# 1. Configure deployment
configure(
    action="configure",
    entrypoint="agent.py",
    agent_name="research-agent",
    memory_mode="STM_AND_LTM"
)

# 2. Create memory
memory(
    action="create",
    name="research-memory",
    strategies=[{
        "semanticMemoryStrategy": {
            "name": "Facts",
            "namespaces": ["/users/{actorId}/facts"]
        }
    }],
    wait_for_active=True
)

# 3. Deploy to AWS
result = launch(
    action="launch",
    agent_name="research-agent",
    mode="codebuild"
)

agent_arn = result["agent_arn"]

# 4. Check deployment status
status(agent_name="research-agent")

# 5. Invoke agent
invoke(
    agent_arn=agent_arn,
    payload='{"prompt": "Analyze this data"}',
    session_id="session-abc-123-very-long-session-id-here"
)

# 6. Check logs
logs(agent_name="research-agent", action="recent", limit=50)

---

Use with Strands Agents

Enable autonomous deployment by giving agents these tools:

from strands import Agent
from strands_agentcore_tools import (
    configure, launch, invoke, status, logs
)

agent = Agent(
    tools=[configure, launch, invoke, status, logs],
    system_prompt="You can deploy yourself to AWS AgentCore."
)

# Agent deploys itself
response = agent("""
Deploy yourself to AWS:
1. Configure deployment
2. Launch to production
3. Check status
4. Show logs
""")

Agent will autonomously:

• ✅ Generate Dockerfile and config
• ✅ Trigger CodeBuild deployment
• ✅ Monitor deployment status
• ✅ Validate with logs
• ✅ Report results

---

Configuration File

.bedrock_agentcore.yaml (auto-generated by configure):

agents:
  my-agent:
    name: my-agent
    entrypoint: agent.py
    platform: linux/arm64
    aws:
      execution_role: arn:aws:iam::123:role/AgentCoreRuntime-us-west-2-abc
      region: us-west-2
      ecr_repository: 123.dkr.ecr.us-west-2.amazonaws.com/bedrock-agentcore-my-agent
    memory:
      mode: STM_AND_LTM
      memory_id: memory-abc123
    bedrock_agentcore:
      agent_id: my-agent-abc123
      agent_arn: arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc123

---

IAM Permissions

Runtime Execution Role (auto-created by configure):

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream",
      "logs:CreateLogGroup",
      "logs:CreateLogStream",
      "logs:PutLogEvents",
      "bedrock-agentcore:CreateMemoryEvent",
      "bedrock-agentcore:RetrieveMemoryRecords",
      "ecr:GetAuthorizationToken",
      "ecr:BatchGetImage",
      "ecr:GetDownloadUrlForLayer",
      "xray:PutTraceSegments",
      "xray:PutTelemetryRecords"
    ],
    "Resource": "*"
  }]
}

Your IAM User/Role (attach manually):

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "bedrock-agentcore-control:*",
      "bedrock-agentcore:*",
      "ecr:*",
      "codebuild:*",
      "iam:CreateRole",
      "iam:GetRole",
      "iam:PassRole",
      "iam:PutRolePolicy",
      "logs:*",
      "s3:PutObject",
      "s3:GetObject"
    ],
    "Resource": "*"
  }]
}

---

Error Handling

All tools return structured responses:

# Success
{
    "status": "success",
    "agent_arn": "arn:aws:bedrock-agentcore:us-west-2:123:runtime/my-agent-abc",
    "message": "Agent deployed successfully"
}

# Error
{
    "status": "error",
    "error": "ConflictException",
    "message": "Agent already exists. Use auto_update_on_conflict=True"
}

Common Errors:

• ConflictException - Agent exists, use auto_update_on_conflict=True in launch()
• ValidationException - Invalid parameters (check agent_name format)
• AccessDeniedException - IAM permissions missing
• ResourceNotFoundException - Agent/memory not found

---

Observability

View logs programmatically:

from strands_agentcore_tools import logs

# Recent logs
logs(agent_name="my-agent", action="recent", limit=50)

# Search for errors
logs(agent_name="my-agent", action="search", filter_pattern="ERROR")

# Tail in real-time
logs(agent_name="my-agent", action="tail")

Add ADOT instrumentation for tracing:

# In Dockerfile (auto-generated by configure)
RUN pip install aws-opentelemetry-distro>=0.10.1
CMD ["opentelemetry-instrument", "python", "agent.py"]

CloudWatch Logs Location:

/aws/bedrock-agentcore/runtimes/<agent_id>-<endpoint>

---

Development

Local Testing:

# Install package locally
pip install -e .

# Run tests
pytest tests/

# Format code
black strands_agentcore_tools/

Project Structure:

strands-agentcore-tools/
├── strands_agentcore_tools/
│   ├── __init__.py
│   ├── configure.py        # Deployment configuration
│   ├── launch.py           # CodeBuild deployment
│   ├── invoke.py           # Agent invocation
│   ├── agents.py           # Agent discovery
│   ├── status.py           # Health checks
│   ├── logs.py             # CloudWatch logs
│   ├── memory.py           # Memory management
│   ├── identity.py         # OAuth providers
│   └── session.py          # Session control
├── setup.py
├── pyproject.toml
└── README.md

---

Resources

Documentation:

• Strands Agents Docs
• AWS Bedrock AgentCore Docs
• AgentCore API Reference

Related Projects:

• bedrock-agentcore-starter-toolkit - Official AWS toolkit
• strands-agents - Strands SDK

---

License

Apache 2.0 License - see LICENSE file for details.

---

Contributing

Issues and PRs welcome at github.com/cagataycali/strands-agentcore-tools