minorun365/my-claude-code-settings/kb-agentcore-identity

# ❌ NG: デコレータをスタック → パラメータ干渉
@tool
@requires_access_token(provider_name="my-provider", scopes=[...], auth_flow="USER_FEDERATION")
async def get_pages(*, access_token: str):
    ...
# ✅ OK: トークン取得とツールを分離
@requires_access_token(provider_name="my-provider", scopes=[...], auth_flow="USER_FEDERATION",
    on_auth_url=lambda url: print(f"認可URL: {url}"),
    callback_url="http://localhost:9090/oauth2/callback")
def get_token(access_token: str = ""):
    return access_token
@tool
def get_pages():
    """外部APIのページ一覧を取得する"""
    token = get_token()
    response = httpx.get("https://api.example.com/pages",
        headers={"Authorization": f"Bearer {token}"})
    return response.json()

@tool
def add_calendar_event(
    summary: str,
    start_datetime: str,
    end_datetime: str,
    description: str = "",
):
    """Google Calendarに予定を追加する。"""
    @requires_access_token(
        provider_name=PROVIDER_NAME,
        scopes=["https://www.googleapis.com/auth/calendar"],
        auth_flow="USER_FEDERATION",
        on_auth_url=lambda url: print(f"認証URL: {url}"),
        callback_url="http://localhost:9090/oauth2/callback",
    )
    def call_api(access_token: str = ""):
        event = {
            "summary": summary,
            "start": {"dateTime": start_datetime, "timeZone": "Asia/Tokyo"},
            "end": {"dateTime": end_datetime, "timeZone": "Asia/Tokyo"},
        }
        resp = requests.post(
            "https://www.googleapis.com/calendar/v3/calendars/primary/events",
            headers={"Authorization": f"Bearer {access_token}"},
            json=event,
        )
        return resp.json()
    result = call_api()
    if "error" in result:
        return f"失敗: {result['error']['message']}"
    return f"登録完了: {result.get('summary')}"

# callback_server.py（FastAPI で localhost:9090 に立てる）
from fastapi import FastAPI
from bedrock_agentcore.services.identity import IdentityClient, UserIdIdentifier
app = FastAPI()
identity_client = IdentityClient(region="us-east-1")
@app.get("/oauth2/callback")
async def handle_callback(session_id: str):
    identity_client.complete_resource_token_auth(
        session_uri=session_id,
        user_identifier=UserIdIdentifier(user_id="<user_id>"),
    )
    return {"status": "success"}

# エージェント側：callback_url を localhost に向ける
@requires_access_token(
    provider_name="my-provider",
    scopes=["read:confluence-content.all", "offline_access"],
    auth_flow="USER_FEDERATION",
    on_auth_url=lambda url: print(f"BROWSER: {url}"),
    callback_url="http://localhost:9090/oauth2/callback",
)
def get_token(access_token: str = ""):
    return access_token

# ❌ NG: モジュールレベルで実行 → 30秒タイムアウト
agent = Agent(tools=[my_tool])
agent("処理を実行して")
# ✅ OK: BedrockAgentCoreApp + @app.entrypoint で invoke 時に実行
app = BedrockAgentCoreApp()
@app.entrypoint
def handle_request(request, context=None):
    agent = Agent(tools=[my_tool])
    result = agent(request.get("prompt", ""))
    return {"response": str(result)}
if __name__ == "__main__":
    app.run()

{
  "Sid": "BedrockAgentCoreIdentityGetResourceOauth2Token",
  "Effect": "Allow",
  "Action": ["bedrock-agentcore:GetResourceOauth2Token"],
  "Resource": [
    "arn:aws:bedrock-agentcore:REGION:ACCOUNT:token-vault/default",
    "arn:aws:bedrock-agentcore:REGION:ACCOUNT:token-vault/default/oauth2credentialprovider/*",
    "arn:aws:bedrock-agentcore:REGION:ACCOUNT:workload-identity-directory/default",
    "arn:aws:bedrock-agentcore:REGION:ACCOUNT:workload-identity-directory/default/workload-identity/*"
  ]
}

{"message":"2 validation errors detected: Value at 'authorizationCode' failed to satisfy constraint: Member must not be null; Value at 'state' failed to satisfy constraint: Member must not be null"}

callback_url="http://localhost:9090/oauth2/callback"
identity_client.complete_resource_token_auth(
    session_uri=session_id,
    user_identifier=UserIdIdentifier(user_id="<user_id>"),
)