Kuso Entertainment MCP

kuso娱乐mcp

这对特工来说是很棒的MCP，对人类来说是kuso

概述

此示例演示了 使用AgentCore网关构建的安全MCP服务器 两者都有 入站OAuth身份验证 （用户身份）和 出站OAuth授权 （API访问）。

AgentCore网关充当MCP服务器，处理：

• 入站身份验证: 验证 用户通过谷歌OAuth（通过Cognito联盟）→ 识别用户为世界卫生组织
• 出站身份验证: 授权 用户通过代币保险库访问API→ 授权他们可以访问的内容

用例：构建安全的MCP服务器，使用授权工具代表经过身份验证的用户访问第三方API。

建筑

kuso_agent

为什么我们需要Cognito？

AgentCore身份提供程序具有不同的功能：

AgentCore中的Google OAuth提供程序已设计完成 仅用于出站资源访问，不用于入站用户身份验证(AWS文档).

解决方案：带有Google Federation的Cognito允许Google登录进行入站身份验证，同时保留Google用户ID以进行出站令牌查找。

先决条件

• 具有AgentCore访问权限的AWS帐户
• Google OAuth应用程序凭据(https://console.cloud.google.com/apis/credentials)

- 启用YouTube数据API v3

• Python 3.10+
• 已配置AWS凭据
• uv 用于依赖关系管理

快速开始

cd oauth-gateway-from-agent
cp .env.example .env
# Edit .env with GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET
uv run python construct.py
# Register callback URLs in Google OAuth App (shown after construct.py)
uv run python main.py  # First run: initializes workload identity
uv run python main.py  # Second run: completes OAuth and runs agent

注： 首次运行初始化 .agentcore.json 随着 user_id 并退出。第二次运行使用此 user_id 在用于正确OAuth会话绑定的回调URL中。

演示流程

sequenceDiagram
    participant User as main.py (Agent)
    participant Identity as AgentCore Identity
    participant Cognito as Cognito User Pool
    participant Google as Google OAuth
    participant Gateway as Gateway (MCP Server)
    participant Vault as Token Vault
    participant Callback as Callback Server
    participant YouTube as YouTube API

    Note over User,YouTube: Phase 1: Inbound Authentication (Cognito + Google Federation)
    User->>Identity: @requires_access_token (openid, email, profile)
    Identity->>Cognito: Redirect to Hosted UI
    Cognito->>Google: Redirect to Google sign-in
    Google-->>User: on_auth_url callback (browser opens)
    User->>Google: Sign in with Google
    Google-->>Cognito: Authorization code
    Cognito->>Cognito: Exchange code, create user
    Cognito-->>Identity: Cognito JWT (with identities claim)
    Identity-->>User: access_token (Cognito JWT)

    Note over User,YouTube: Phase 2: Connect to Gateway as MCP Client
    User->>Gateway: MCP initialize + Bearer Cognito JWT
    Gateway->>Gateway: Validate Cognito JWT ✓
    Gateway-->>User: MCP initialized, tools available

    Note over User,YouTube: Phase 3: First Tool Call → Outbound Auth (3LO)
    User->>Gateway: tools/call (list_channels)
    Gateway->>Vault: Get YouTube token for Cognito user X
    Vault-->>Gateway: No token (authorizationUrl + session_id)
    Gateway-->>User: 401 + authorizationUrl
    User->>User: Store (session_id → Cognito JWT) in DynamoDB
    
    User->>Google: Authorize YouTube scope (browser)
    Google-->>Callback: Redirect with session_id
    Callback->>Callback: Retrieve Cognito JWT from DynamoDB
    Callback->>Vault: CompleteResourceTokenAuth(session_id, userToken=JWT)
    Vault-->>Callback: Token stored for user
    Callback-->>User: "Authorization Complete!"

    Note over User,YouTube: Phase 4: Subsequent Calls → Token from Vault
    User->>Gateway: tools/call (list_channels)
    Gateway->>Vault: Get YouTube token for Cognito user X
    Vault-->>Gateway: YouTube API token ✓
    Gateway->>YouTube: GET /youtube/v3/channels
    YouTube-->>Gateway: Channel data
    Gateway-->>User: Tool result (channels)

施工流程

sequenceDiagram
    participant Dev as Developer
    participant Script as construct.py

    box CloudFormation Stack
        participant CF as CloudFront + Lambda
        participant Cognito as Cognito User Pool
        participant DB as DynamoDB
    end

    participant AC as AgentCore

    Dev->>Script: uv run python construct.py

    Note over Script,DB: Step 1: Deploy CloudFormation Stack
    Script->>CF: Create CloudFront + Lambda callback server
    CF-->>Script: callback_url
    Script->>Cognito: Create User Pool + Google IdP
    Cognito-->>Script: user_pool_id, client_id, discovery_url
    Script->>DB: Create session storage table
    DB-->>Script: table_name

    Note over Script,AC: Step 2: AgentCore Identity
    Script->>AC: Create Inbound OAuth Provider (Cognito)
    AC-->>Script: inbound_provider_arn
    Script->>AC: Create Outbound OAuth Provider (Google/YouTube)
    AC-->>Script: outbound_provider_arn

    Note over Script,AC: Step 3: AgentCore Gateway
    Script->>AC: Create Gateway (CUSTOM_JWT validates Cognito JWT)
    AC-->>Script: gateway_id
    Script->>AC: Create Gateway Target (YouTube API)
    AC-->>Script: target_id
    Script->>AC: Update Workload Identity (register callback URL)
    AC-->>Script: workload_identity_updated

    Script->>Dev: Save config.json
    Script->>Dev: Display callback URLs to register in Google OAuth App

规格

组件责任

Cognito用户池（谷歌联盟）

Cognito充当谷歌和Gateway之间的身份中介：

智威汤逊声明（谷歌登录后）：

{
  "sub": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
  "email": "user@example.com",
  "cognito:username": "Google_123456789",
  "identities": [
    {
      "userId": "123456789",
      "providerName": "Google",
      "providerType": "Google",
      "issuer": null,
      "primary": "true",
      "dateCreated": "1642699117273"
    }
  ],
  "iss": "https://cognito-idp..amazonaws.com/",
  "aud": "",
  "token_use": "id"
}

这 identities 声明包含Google用户ID(userId)，但系统使用Cognito用户ID(sub 用于令牌库中的令牌存储和检索。

为什么选择Cognito sub 而不是谷歌 userId?

这 sub （主体）索赔是 标准JWT索赔 其在发行系统（Cognito）内唯一标识用户。AgentCore的令牌库使用JWT sub 声明为用户标识符，而不是像这样的自定义声明 identities.

当 CompleteResourceTokenAuth 被称为：

1. 试剂核心提取物 sub JWT的索赔（例如。， "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
2. 将令牌存储为： (Workload Identity + sub claim) → YouTube token
3. 使用相同的方法检索令牌 sub 后续请求中的索赔

使用来自的Google用户ID identities 索赔将造成不匹配-AgentCore仍将使用 sub 内部索赔，导致令牌存储/检索失败。

OAuth回调服务器（CloudFront+Lambda）

回调服务器处理两者的OAuth会话绑定 入站 和 出站 身份验证流程。

两条回调路径：

为什么有不同的方法？

这 complete_resource_token_auth API需要 userIdentifier 将OAuth会话绑定到特定用户。这可以防止授权URL转发攻击。

入站流量(/inbound?user_id=xxx&session_id=yyy):

1. main.py 读取 user_id 从 .agentcore.json （由IdentityClient创建）
2. 追加 user_id 回调URL： /inbound?user_id=xxx
3. 用户完成Cognito/Google登录
4. Lambda同时接收回调 session_id 和 user_id
5. Lambda调用 complete_resource_token_auth(sessionUri, userIdentifier={userId: user_id})

出站流量(/?session_id=yyy):

1. main.py 商店 (session_id → encrypted_user_token) 重定向前在DynamoDB中
2. 用户完成YouTube授权
3. Lambda检索和解密 user_token 来自DynamoDB
4. Lambda调用 complete_resource_token_auth(sessionUri, userIdentifier={userToken: jwt})

首次运行初始化：

在第一次运行时， .agentcore.json 还不存在，所以 user_id 不可用。脚本检测到这一点并初始化工作负载标识：

if "user_id=" not in get_inbound_callback_url():
    print("Initializing workload identity...")
    # Creates .agentcore.json with workload_identity_name and user_id
    asyncio.run(_get_workload_access_token(IdentityClient(region=REGION)))
    print("Initialized. Please run again.")
    exit(0)

架构：

• CloudFront分布：具有DDoS保护和WAF功能的公共端点
• 匿名函数：处理OAuth回调并完成会话绑定
• DynamoDB表：使用TTL存储会话数据以进行自动清理（仅限出站）
• 源访问控制（OAC）：仅限制Lambda访问CloudFront

关键实施细节：

• Lambda需要通过Layer使用最新的boto3（内置boto3缺少 bedrock-agentcore 服务）
• 工作负载标识必须在中注册两个回调URL allowedResourceOauth2ReturnUrls:

- https://cloudfront.net/ （出境） - https://cloudfront.net/inbound （入境）

• 入站使用 userId （稳定，8字符十六进制），出站使用 userToken （完整的JWT）
• Lambda需要 secretsmanager:GetSecretValue 许可（需要 CompleteResourceTokenAuth)

OAuth提供者

main.py（Strands代理）

关于的注释 @requires_access_token 对于入站身份验证： 当 @requires_access_token 通常用于出站OAuth（获取API令牌），我们在这里利用它 入站身份验证 简化OAuth流程。装饰器处理基于浏览器的OAuth舞蹈，并返回一个Cognito JWT（与谷歌联盟），该JWT：

1. 通过Google登录对用户进行身份验证（他们是世界卫生组织）
2. 由网关的CUSTOM_JWT授权人验证
3. 包含 identities 使用谷歌用户ID进行索赔，但系统使用Cognito用户ID(sub claim）用于令牌操作

安全：令牌存储TTL 加密令牌存储的5分钟TTL仅影响 初始OAuth授权流，不是正在进行的YouTube API访问：

• OAuth设置期间：临时存储用于会话绑定的用户令牌（5分钟）
• OAuth完成后：令牌已删除，Gateway直接管理YouTube令牌
• 持续使用：用户可以访问YouTube API数小时/数天，无需重新授权
• 谷歌控制过期：Gateway自动刷新YouTube令牌（通常为1小时访问令牌）
• 只有在以下情况下才需要重新授权：谷歌撤销访问权限或刷新令牌过期（数月不活动）

from bedrock_agentcore.identity import requires_access_token
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp import MCPClient

@requires_access_token(
    provider_name=INBOUND_PROVIDER_NAME,  # Cognito with Google federation
    scopes=["openid", "email", "profile"],
    auth_flow="USER_FEDERATION",
    on_auth_url=lambda url: webbrowser.open(url),  # Opens browser for Google sign-in
)
def run_agent(*, access_token: str):
    """Run agent with Gateway as MCP server.
    
    The access_token is a Cognito JWT (Google federated) containing:
    - sub: Cognito user ID
    - identities: [{"providerName": "Google", "userId": ""}]
    
    Gateway validates this JWT. The agent extracts Cognito user ID (`sub` claim)
    for token storage and session binding operations.
    """
    mcp_client = MCPClient(
        lambda: streamablehttp_client(
            GATEWAY_ENDPOINT,
            headers={"Authorization": f"Bearer {access_token}"}
        )
    )
    with mcp_client:
        tools = mcp_client.list_tools_sync()
        agent = Agent(tools=tools)
        response = agent("List my YouTube channels")
        print(response)

AgentCore网关

AgentCore网关充当 MCP服务器 它通过内置的身份验证和授权将您的代理连接到外部API。

Gateway的角色

┌─────────────────────────────────────────────────────────────────────┐
│                      AgentCore Gateway                              │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  1. INBOUND AUTH: "Who is this user?"                               │
│     - Validates JWT from Cognito (CUSTOM_JWT authorizer)            │
│     - Extracts user identity (sub claim, identities claim)          │
│                                                                     │
│  2. WORKLOAD IDENTITY: "Who is this application?"                   │
│     - Gateway has its own identity in AgentCore                     │
│     - Scopes user tokens: (Gateway + User) → unique token entry     │
│     - Defines trusted callback URLs for OAuth flows                 │
│                                                                     │
│  3. OUTBOUND AUTH: "What can this user access?"                     │
│     - Retrieves user's API tokens from Token Vault                  │
│     - Triggers OAuth flow if token not found (elicitation)          │
│     - Injects tokens into requests to external APIs                 │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

工作负载标识

创建网关时，AgentCore会自动创建 工作负载标识 将其视为Gateway在AgentCore世界中的“企业形象”：

Workload Identity = "mcp-oauth-gateway-gateway-xyz"
        │
        ├── Represents THIS Gateway application
        │
        ├── Token Vault entries are scoped to (Workload Identity + User)
        │   └── Same user through different Gateways = separate tokens
        │
        └── allowedResourceOauth2ReturnUrls
            └── Trusted callback URLs for OAuth session binding

为什么工作负载身份对OAuth很重要：

把它想象成一家处理员工请求的公司：

Workload Identity = "Acme Corp Gateway" (the company)
User (JWT)        = Employee with ID badge
Token Vault       = Secure filing cabinet
session_id        = Request form number

Scenario: Employee B needs YouTube API access

1. Employee B (JWT) enters Acme Corp building
2. Security validates: "Is this badge from Acme Corp?" ✓
3. Employee B: "I need YouTube access"
4. Acme Corp checks filing cabinet: "Acme Corp → Employee B → YouTube... not found"
5. Acme Corp: "Fill out authorization form #session_id, go to Google to approve"
6. Employee B completes OAuth at Google
7. Google sends Employee B back to Acme Corp's registered address
   (allowedResourceOauth2ReturnUrls = company's trusted mailroom)
8. Acme Corp mailroom verifies: "Is this really Employee B completing THEIR form?"
   → Calls CompleteResourceTokenAuth(session_id, Employee B's badge)
9. AgentCore stores: "Acme Corp → Employee B → YouTube → [token]"

Why the registered address matters:
- If a stranger intercepts the form and tries to complete it at a fake address,
  AgentCore rejects it: "That's not Acme Corp's registered mailroom!"
- Only callbacks from allowedResourceOauth2ReturnUrls are trusted

# Register callback URL with Workload Identity
client.update_workload_identity(
    name=gateway_id,  # Workload Identity name = Gateway ID
    allowedResourceOauth2ReturnUrls=["https://your-callback.cloudfront.net/"]
)

网关配置

# Inbound: Cognito JWT validation
authorizerType="CUSTOM_JWT"
authorizerConfiguration={
    "customJWTAuthorizer": {
        "discoveryUrl": cognito_discovery_url,
        "allowedClients": [cognito_client_id]
    }
}

# Outbound: OAuth with session binding for 3LO
credentialProviderConfigurations=[{
    "credentialProviderType": "OAUTH",
    "credentialProvider": {
        "oauthCredentialProvider": {
            "providerArn": outbound_provider_arn,
            "grantType": "AUTHORIZATION_CODE",
            "defaultReturnUrl": callback_url,  # Where to redirect after OAuth
            "scopes": ["https://www.googleapis.com/auth/youtube.readonly"]
        }
    }
}]

OAuth会话绑定流

1. User requests YouTube data → Gateway checks Token Vault → No token found
2. Gateway returns elicitation with auth URL containing session_id
3. User completes OAuth at Google
4. Google redirects to AgentCore → AgentCore redirects to YOUR callback (defaultReturnUrl)
5. Your callback calls CompleteResourceTokenAuth(session_id, userToken)
   - session_id: identifies this authorization attempt
   - userToken: the inbound JWT proving user identity
6. AgentCore stores token in Vault: (Workload Identity + User) → YouTube token
7. Subsequent requests use stored token automatically

安全考虑

• 令牌隔离：每个用户的Google令牌单独存储在令牌库中
• 身份绑定：令牌检索需要从JWT中匹配用户ID
• 范围分离：入站（标识）和出站（API）使用不同的作用域
• 无代币敞口：Google API令牌从未发送到客户端

故障排除

参考文献

Agent核心文档

• AgentCore网关概述
• AgentCore身份-令牌库
• OAuth 2.0凭据提供程序

谷歌OAuth

• 创建OAuth 2.0客户端ID
• OAuth 2.0适用于Google API的范围
• YouTube数据API

MCP（模型上下文协议）

• MCP规范
• MCP身份验证