<a href="https://github.com/aws/bedrock-agentcore-sdk-python/blob/main/LICENSE"><img alt="License" src="https://img.shields.io/github/license/aws/bedrock-agentcore-sdk-python"/></a><a href="https://pypi.org/project/bedrock-agentcore"><img alt="PyPI version" src="https://img.shields.io/pypi/v/bedrock-agentcore"/></a><a href="https://python.org"><img alt="Python versions" src="https://img.shields.io/pypi/pyversions/bedrock-agentcore"/></a>

# 基础安装pip install bedrock-agentcore
# 包含Strands代理集成pip install 'bedrock-agentcore[strands-agents]'
# 开发模式安装git clone https://github.com/aws/bedrock-agentcore-sdk-python.gitcd bedrock-agentcore-sdk-pythonuv syncsource .venv/bin/activate

# 配置AWS凭证export AWS_ACCESS_KEY_ID=your_access_keyexport AWS_SECRET_ACCESS_KEY=your_secret_keyexport AWS_REGION=us-west-2
# 可选：自定义端点（用于测试环境）export BEDROCK_AGENTCORE_DP_ENDPOINT=your_data_plane_endpointexport BEDROCK_AGENTCORE_CP_ENDPOINT=your_control_plane_endpoint

from bedrock_agentcore.runtime import BedrockAgentCoreAppfrom bedrock_agentcore.runtime import RequestContext
# 创建应用实例app = BedrockAgentCoreApp()
@app.run()async def my_agent_handler(context: RequestContext, input_text: str) -> str:    """简单的AI代理处理器"""    return f"处理后的响应: {input_text}"
# 启动应用if __name__ == "__main__":    app.start()

from bedrock_agentcore.memory import MemorySessionManagerfrom bedrock_agentcore.memory.constants import MessageRole
# 初始化会话管理器manager = MemorySessionManager(    memory_id="my-memory-123",    region_name="us-east-1")
# 创建会话session = manager.create_session(    actor_id="user-123",    session_id="conversation-456")
# 添加对话轮次await session.add_turn(    user_message="今天天气怎么样？",    assistant_message="今天天气晴朗，温度适宜。")
# 检索相关记忆records = await session.retrieve_memory_records(    query="天气信息",    top_k=5)

from bedrock_agentcore.auth import requires_access_token
@requires_access_token(    provider_name="my-oauth-provider",    scopes=["read:data", "write:data"],    auth_flow="USER_FEDERATION")async def protected_operation(access_token: str, data: dict) -> dict:    """需要OAuth2访问令牌的保护操作"""    # 使用access_token调用受保护的API    return {"status": "success", "data": data}

from bedrock_agentcore.tools import browser_session
async def web_automation_example():    """浏览器自动化示例"""    async with browser_session(region="us-west-2") as browser:        # 导航到网页        await browser.goto("https://example.com")                # 执行页面操作        content = await browser.get_content()        return f"页面内容: {content}"

"""BedrockAgentCore Runtime SDK - 构建和部署AI代理的Python SDK"""
from .runtime import BedrockAgentCoreApp, BedrockAgentCoreContext, RequestContextfrom .runtime.models import PingStatus
__all__ = [    "BedrockAgentCoreApp",    "RequestContext",     "BedrockAgentCoreContext",    "PingStatus",]

"""AgentCore Memory SDK - 记忆操作的高级客户端"""
class MemoryClient:    """Bedrock AgentCore Memory的高级客户端，提供基本操作"""        # AgentCore Memory数据平面方法    _ALLOWED_GMDP_METHODS = {        "retrieve_memory_records",        "get_memory_record",         "delete_memory_record",        "list_memory_records",        "create_event",        "get_event",        "delete_event",         "list_events",    }        # AgentCore Memory控制平面方法      _ALLOWED_GMCP_METHODS = {        "create_memory",        "get_memory",        "list_memories",        "update_memory",        "delete_memory",        "list_memory_strategies",    }
    def __init__(self, region_name: Optional[str] = None):        """初始化Memory客户端"""        self.region_name = region_name or boto3.Session().region_name or "us-west-2"                self.gmcp_client = boto3.client("bedrock-agentcore-control", region_name=self.region_name)        self.gmdp_client = boto3.client("bedrock-agentcore", region_name=self.region_name)

"""Bedrock AgentCore SDK的身份验证装饰器和工具"""
def requires_access_token(    *,    provider_name: str,    into: str = "access_token",     scopes: List[str],    on_auth_url: Optional[Callable[[str], Any]] = None,    auth_flow: Literal["M2M", "USER_FEDERATION"],    callback_url: Optional[str] = None,    force_authentication: bool = False,    token_poller: Optional[TokenPoller] = None,) -> Callable:    """在调用装饰函数之前获取OAuth2访问令牌的装饰器"""        def decorator(func: Callable) -> Callable:        client = IdentityClient(_get_region())
        async def _get_token() -> str:            """通用的令牌获取逻辑"""            return await client.get_token(                provider_name=provider_name,                agent_identity_token=await _get_workload_access_token(client),                scopes=scopes,                on_auth_url=on_auth_url,                auth_flow=auth_flow,                callback_url=callback_url,                force_authentication=force_authentication            )

"""包含AgentCore Memory交互的会话管理类"""
class MemorySessionManager:    """管理AWS Bedrock AgentCore的会话和记忆操作"""        def __init__(self, memory_id: str, region_name: str = "us-west-2"):        """初始化MemorySessionManager"""        self.memory_id = memory_id        self.region_name = region_name        self.client = MemoryClient(region_name=region_name)            async def create_session(self, actor_id: str, session_id: str) -> "MemorySession":        """为指定参与者和会话创建新的MemorySession"""        return MemorySession(            memory_id=self.memory_id,            actor_id=actor_id,            session_id=session_id,            client=self.client        )

"""与浏览器沙箱服务交互的客户端"""
class BrowserClient:    """与AWS浏览器沙箱服务交互的客户端"""        def __init__(self, region: str) -> None:        """为指定AWS区域初始化浏览器客户端"""        self.region = region        self.data_plane_service_name = "bedrock-agentcore"        self.client = boto3.client(            self.data_plane_service_name,             region_name=region,             endpoint_url=get_data_plane_endpoint(region)        )        self._identifier = None        self._session_id = None