Obtain an OAuth Access Token

Workflow

Agent Identity currently supports obtaining OAuth Access Tokens for resources through user delegation.

User delegation: When an agent completes inbound authentication and requests an Access Token from Agent Identity, Agent Identity returns an OAuth authorization URL to the agent. The agent then renders this URL to the user for authorization. After the user grants authorization, the agent obtains an Access Token from the configured OAuth credential provider on behalf of the user. The agent then stores the token and returns it.

The following example describes the workflow for an agent obtaining an OAuth access token on behalf of a user for a downstream service, such as DingTalk, and completing resource access in a user delegation scenario that uses the OAuth 2.0 authorization code grant flow:

1. Invoke agent. A user logs on to the application and invokes the agent. The agent analyzes the user's request, for example, "Help me write this content to my DingTalk document", and determines that it needs to invoke the DingTalk Open Service. Then, the agent uses the Agent Identity SDK to request an OAuth access token from the DingTalk credential provider through Alibaba Cloud Agent Identity.

   Note

   The state parameter is an opaque string generated by the application to prevent Cross-Site Request Forgery (CSRF) attacks. If you pass the state parameter when you request an OAuth Access Token, Agent Identity returns the state parameter unchanged to the application's callback address after the user completes authorization.

2. Generate authorization URL. The Agent Identity SDK determines the inbound authentication method based on the user context passed by the agent. For example, if the user context contains a user's JSON Web Token (JWT), which is an ID Token, the SDK invokes the GetWorkloadAccessForJWT interface to exchange the user's JWT for a Workload Access Token. Then, the Agent Identity SDK uses the obtained Workload Access Token to request an OAuth Access Token from Agent Identity. Agent Identity generates the OAuth authorization URL for the DingTalk credential provider and returns it to the application.

   Note

   Agent Identity returns a session_uri along with the OAuth authorization URL to the agent. Agent Identity uses the session_uri to associate the user information passed during the agent's inbound authentication with the current session. In other words, the session_uri is bound to the user who initially invoked the agent.

   The session_uri in each generated OAuth authorization URL is unique, and the entire OAuth authorization URL can be used only once.

3. Authorize and obtain an Access Token. The application renders the OAuth authorization URL to the user for authorization. After the user completes the DingTalk logon and authorization, the application or browser submits the OAuth authorization code to Agent Identity through redirection. After Agent Identity receives the authorization code, it returns the session_uri and state parameters to the application's callback address. The application verifies the state parameter and the current user's logon information, and then invokes the CompleteResourceTokenAuth interface of Agent Identity to continue the process of obtaining an Access Token. For more information, see Session binding.

4. Re-invoke the agent to obtain the Access Token. After the application successfully invokes the CompleteResourceTokenAuth interface, the agent can obtain the Access Token that is associated with the initial requester from Agent Identity.

5. Access the resource. The agent uses the obtained Access Token to access the resource.

Session binding

A user might obtain an OAuth authorization URL, send it to another user, and delegate the authentication. This would result in the initial user obtaining the other user's Access Token and access privileges. The session binding feature of Agent Identity ensures that the user who initially invoked the agent and obtained the OAuth authorization URL is the same user who completes the OAuth authorization.

The implementation is as follows:

1. The application must register a callback address with Agent Identity. This is set when you configure Workload Identity.

2. Agent Identity does not immediately request an Access Token after it receives the OAuth authorization code. Instead, it returns the session_uri, which is bound to the initial invoker, and the state parameter, which is passed by the application when it invokes the agent, to the application's callback address.

3. The application must add processing code at the listening callback address. This code confirms that the user who completed the OAuth authorization is the same as the initial invoker by checking the state parameter and the cookie information in the request. After confirmation, the application invokes the CompleteResourceTokenAuth interface of Agent Identity and passes the session_uri and the current user's logon information, such as user_id or JWT, to continue obtaining the OAuth Access Token.

4. Agent Identity compares the passed user information, which is obtained from the user ID or user token, with the initial invoker information in the session_uri. If they match, Agent Identity continues to obtain the Access Token using the OAuth authorization code and returns a success message to the application.

The following Python example code shows how to handle the callback address:

@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 Agent Identity calls back to the backend service, 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
    # 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. Deny confirmation in this case.
    if user_session_map.get(state) is None:
        raise HTTPException(status_code=400, detail="Invalid state")
    # Try to get initial user's session id 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 parse initial user token 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 storage and refresh

Agent Identity encrypts and stores the obtained OAuth Access Token and Refresh Token, if one is returned by the credential provider, in the Token Vault. When the agent requests an Access Token again, Agent Identity prioritizes returning a valid Access Token from the Token Vault. If the Access Token has expired, Agent Identity uses the Refresh Token to request a new Access Token from the credential provider and returns the new token. Users do not need to perform OAuth authentication and authorization again before the Refresh Token expires.

If the Refresh Token has expired or become invalid, Agent Identity returns the OAuth authorization URL again to require the user to re-authenticate and re-authorize.

Usage

After you create an OAuth credential provider in Agent Identity, you can use the Agent Identity SDK in the agent to obtain an OAuth Access Token. Before a function that needs an Access Token to invoke other services, add the @requires_access_token annotation and pass the following parameters. The Agent Identity SDK automatically obtains a Workload Access Token for you and uses it to request an OAuth Access Token.

• OAuth credential provider name: The name of the OAuth credential provider that you configured in Agent Identity. For more information about configuration methods, see Manage OAuth credential providers.

• User context information (optional): Contains user information for using the agent, such as a user ID and user token (ID Token). The Agent Identity SDK uses the user context information to determine how to obtain the Workload Access Token.

The following Python example code shows how to obtain an OAuth Access Token in a user delegation scenario:

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 is 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)

References

Use agent identity in Alibaba Cloud Model Studio high-code