获取OAuth Access Token

工作流程

Agent Identity当前支持以用户委托访问的方式获取资源的OAuth Access Token。

用户委托访问：Agent（已完成入站认证）向Agent Identity服务请求Access Token时，Agent Identity会向Agent返回OAuth授权URL。Agent将该URL呈现给用户以获得授权。授权完成后，Agent将代表用户向配置的OAuth凭证提供商获取Access Token，然后存储并返回给Agent。

下面以用户委托访问（OAuth2.0 auth code grant流）场景为例，介绍Agent代表用户获取下游服务（例如钉钉）的OAuth access token并完成资源访问的工作流程：

1. 调用Agent。用户登录应用并调用Agent。Agent分析用户请求（例如：“帮我将以下内容写入我的钉钉文档”），判定需要调用钉钉开放服务。随后，Agent通过Agent Identity SDK向阿里云Agent Identity服务请求钉钉凭证提供商的OAuth access token。

   说明

   state是一个由应用产生的不透明状态字符串，用于防止CSRF攻击。如果在请求OAuth Access Token时传入state，则Agent Identity服务会在完成用户授权后将state原样返回到应用的回调地址。

2. 生成授权URL。Agent Identity SDK根据Agent传入的用户上下文，判定完成入站认证的具体方式。例如，若Agent传入的用户上下文包含用户JWT令牌（ID Token），则调用GetWorkloadAccessForJWT接口，通过用户的JWT令牌换取Workload Access Token。随后，Agent Identity SDK使用获取的Workload Access Token向Agent Identity服务请求OAuth Access Token。Agent Identity服务生成钉钉凭证提供商对应的OAuth授权URL，并最终返回给应用。

   说明

   Agent Identity服务会将session_uri连同OAuth授权URL一起返回给Agent。Agent Identity使用session_uri将Agent完成入站认证时传入的用户信息与当前会话绑定。换句话说，session_uri与初始调用Agent的用户绑定。

   每一次产生的OAuth授权URL中所包含的session_uri是唯一的，且整个OAuth授权URL只能被使用一次。

3. 授权并获取Access Token。应用将OAuth授权URL呈现给用户以完成授权。用户完成钉钉登录和授权后，应用或浏览器通过重定向将OAuth授权码提交给Agent Identity服务。Agent Identity服务收到授权码后，将session_uri和state返回到应用的回调地址。应用校验state和当前登录的用户信息，然后调用Agent Identity的CompleteResourceTokenAuth接口以继续获取Access Token的流程。具体请参见会话绑定。

4. 重新调用Agent获取Access Token。应用成功调用CompleteResourceTokenAuth接口后，Agent即可从Agent Identity服务获取与初始请求者关联的Access Token。

5. 访问资源。Agent使用获取的Access Token访问资源。

会话绑定

当用户获取OAuth授权URL后，可能会将其发送给其他用户并委托其完成认证。这会导致初始用户获得其他用户的Access Token和资源访问权限。Agent Identity通过会话绑定功能，确保初始调用Agent并获取OAuth授权URL的用户，与完成OAuth授权的用户是同一人。

具体的实现步骤如下：

1. 应用需向Agent Identity服务注册一个回调地址（在配置Workload Identity时设置）。

2. Agent Identity服务不会在收到OAuth授权码后立即请求Access Token，而是会将session_uri（与初始调用者绑定）和state（应用在对Agent发起调用时传入）返回到应用的回调地址。

3. 应用需在监听的回调地址处添加处理代码，通过判断state和请求中包含的Cookie信息，确认完成OAuth授权的用户与初始调用者是同一人。确认后，调用Agent Identity服务的CompleteResourceTokenAuth接口，并传入session_uri和当前用户登录信息（例如user_id或JWT令牌）以继续OAuth Access Token的获取。

4. Agent Identity会比较传入的用户信息（通过用户ID或用户令牌获取）与session_uri中的初始调用者信息。如果两者匹配，则继续使用OAuth授权码获取Access Token，并向应用返回成功消息。

回调地址处理的示例代码（Python）如下：

@app.get("/callback")
async def callback(session_uri: str, state: str, request: Request):
    """
    Callback interface - Handles authentication callback
    
    Args:

        session_uri (str): Session URI returned by Agent Identity callback, used for client confirmation when obtaining OAuth Token

        state (str): State parameter, which is the chat session ID passed through by the backend service to the Agent. 
                     When the Agent Identity callbacks to the backend service, it's recommended to verify 
                     the state against the caller's identity (usually stored in cookies) to ensure 
                     the OAuth authorizer and initiator are the same user.
        request (Request): HTTP request object
        
    Returns:
        str: Success message
        
    Raises:
        HTTPException: When login session ID is missing or invalid
    """
    # Get login_session_id from cookies
    login_session_id = request.cookies.get("oauthSessionId")
    
    # Check if login_session_id exists
    if not login_session_id:
        raise HTTPException(status_code=400, detail="Missing login_session_id cookie")


    # Check if the state (i.e., the chat session ID passed in when initiating chat) exists
    # Need to verify that its corresponding login session ID matches the session ID in the current caller's cookie
    # Otherwise, the authorization link may have been forwarded to someone else, in which case confirmation should be denied
    if user_session_map.get(state) is None:
        raise HTTPException(status_code=400, detail="Invalid state")
    # Try to get initial user's session id by using state
    state_session_id = user_session_map[state]
    # Compare if current logged in user is the initial user who made agent call
    if state_session_id != login_session_id:
        raise HTTPException(status_code=400, detail="Invalid login_session_id")
    # Try to prase initial user token by using state session id 
    user_identifier = UserTokenIdentifier(user_token=user_token_map[state_session_id])
    try:
        identity_client.complete_resource_token_auth(session_uri=session_uri, user_identifier=user_identifier)
    except Exception as e:
        logger.error(e)
        raise e

    return HTMLResponse(content="""
    <html>
    <head>
        <title>Success</title>
        <script>
            setTimeout(function() {
                window.close();
            }, 3000);
        </script>
    </head>
    <body>
        <h1>Success</h1>
        <p>This window will close automatically in 3 seconds.</p>
    </body>
    </html>
    """, status_code=200)

Token的保存和刷新

Agent Identity将获取的OAuth Access Token和Refresh Token（如凭证提供商有返回）加密存储到Token Vault中。当Agent再次请求Access Token时，Agent Identity会优先返回Token Vault中有效的Access Token。如果Access Token已过期，则使用Refresh Token向凭证提供商请求新的Access Token并返回。在Refresh Token过期前，用户无需再次进行OAuth认证和授权。

如果Refresh Token已过期或失效，Agent Identity将再次返回OAuth授权URL，以要求用户重新完成认证和授权。

使用方式

在Agent Identity中创建OAuth凭证提供商后，即可在Agent中使用Agent Identity SDK获取OAuth Access Token。在需要使用Access Token调用其它服务的函数前，添加@requires_access_token注解，并传入以下参数。Agent Identity SDK会为您自动获取Workload Access Token并以此请求OAuth Access Token。

• OAuth凭证提供商名称：您在Agent Identity中配置的OAuth凭证提供商名称，配置方法参见OAuth凭证提供商管理。

• 用户上下文信息（可选）：包含使用Agent的用户信息如用户ID、用户令牌（ID Token）。Agent Identity SDK会使用用户上下文信息来决定如何获取Workload Access Token。

以下为使用用户委托访问场景获取OAuth Access Token的参考示例代码（Python）：

from agent_identity_python_sdk.identity import requires_access_token

@requires_access_token(
    provider_name="your-oauth-credential-provider-name",
    scopes=["profile", "openid", "aliuid", "/acs/mcp-server"],
    auth_flow="USER_FEDERATION", # User delegated (3LO) auth flow
    on_auth_url=lambda x: print("Copy and paste this authorization url to your browser", x), # prints authorization URL to console
    # force_authentication=True,  # When forced authentication is enabled, a new authorization link will be returned every time
    callback_url="http://127.0.0.1:8080/callback",
    user_info_context=<your-user-identifier-context> # user context contains user token or user id
    # custom_parameters={"param1": "test-param", "param2": "test-param2"} # custom OAuth request params
)
async def access_aliyun_mcp(access_token: str):
    if not access_token:
        raise Exception("Access token is required")
    await call_mcp_server(access_token)

相关文档

在百炼高代码中使用Agent Identity