Amazon Bedrock AgentCore로 300개 앱 마이그레이션

import logging
import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from strands.tools.mcp.mcp_types import MCPClientCredentials

logger = logging.getLogger(__name__)
app = BedrockAgentCoreApp()
REGION = os.environ["AWS_REGION"]

# url+auth lets the SDK run the client_credentials grant and re-mint the
# token on expiry. A statically captured bearer token would go stale.
gateway = MCPClient(
    url=os.environ["GATEWAY_MCP_URL"],
    auth=MCPClientCredentials(
        client_id=os.environ["GATEWAY_CLIENT_ID"],
        client_secret=get_secret("gateway/client_secret"),
        scopes=[os.environ["GATEWAY_SCOPE"]],
    ),
)

model = BedrockModel(
    model_id=os.environ["MODEL_ID"],
    region_name=REGION,
    guardrail_id=os.environ["GUARDRAIL_ID"],
    guardrail_version=os.environ.get("GUARDRAIL_VERSION", "1"),
    guardrail_trace="enabled",
)

@app.entrypoint
def invoke(payload, context):
    prompt = (payload.get("prompt") or "").strip()
    if not prompt:
        return {"status": "error", "error": "missing required field: prompt"}
    try:
        # tools=[gateway]: SDK owns the connection lifecycle and paginates
        # tool discovery, which list_tools_sync() alone does not.
        agent = Agent(
            model=model,
            system_prompt=IAC_AGENT_PROMPT,
            tools=[gateway],
        )
        result = agent(prompt)
        if result.stop_reason == "guardrail_intervened":
            logger.warning("guardrail blocked request, session_id=%s", getattr(context, "session_id", None))
            return {"status": "blocked_by_guardrail"}
        return {"status": "ok", "iac": str(result)}
    except Exception as e:
        logger.exception("invocation failed, session_id=%s", getattr(context, "session_id", None))
        return {"status": "error", "error": str(e)}

if __name__ == "__main__":
    app.run()