Identidade do Agente (Agent Identity):Obter um OAuth Access Token

Fluxo de trabalho

Atualmente, o Agent Identity suporta a obtenção de OAuth Access Tokens para recursos por meio de delegação de usuário.

Delegação de usuário: Quando um agente conclui a autenticação de entrada e solicita um Access Token ao Agent Identity, este retorna uma URL de autorização OAuth para o agente. O agente então exibe essa URL ao usuário para autorização. Após o usuário conceder a autorização, o agente obtém um Access Token do provedor de credenciais OAuth configurado em nome do usuário. Em seguida, o agente armazena o token e o retorna.

O exemplo a seguir mostra como um agente obtém um OAuth Access Token em nome de um usuário para um service downstream, como o DingTalk, usando o fluxo de concessão de código de autorização OAuth 2.0:

1. Invoque o agente. Um usuário faz login na aplicação e invoca o agente. O agente analisa a solicitação do usuário, por exemplo, "Ajude-me a escrever este conteúdo no meu documento do DingTalk", e determina que precisa invocar o DingTalk Open Service. Então, o agente usa o SDK do Agent Identity para solicitar um OAuth Access Token do provedor de credenciais do DingTalk por meio do Alibaba Cloud Agent Identity.

   Nota

   O parâmetro state é uma string opaca gerada pela aplicação para evitar ataques de Cross-Site Request Forgery (CSRF). Se você passar o parâmetro state ao solicitar um OAuth Access Token, o Agent Identity retornará o parâmetro state inalterado para o endereço de callback da aplicação após o usuário concluir a autorização.

2. Gere a URL de autorização. O SDK do Agent Identity determina o método de autenticação de entrada com base no contexto de usuário passado pelo agente. Por exemplo, se o contexto do usuário contiver um JSON Web Token (JWT) do usuário, que é um ID Token, o SDK invoca a interface GetWorkloadAccessForJWT para trocar o JWT do usuário por um Workload Access Token. Em seguida, o SDK do Agent Identity usa o Workload Access Token obtido para solicitar um OAuth Access Token ao Agent Identity. O Agent Identity gera a URL de autorização OAuth para o provedor de credenciais do DingTalk e a retorna para a aplicação.

   Nota

   O Agent Identity retorna uma session_uri junto com a URL de autorização OAuth para o agente. O Agent Identity usa a session_uri para associar as informações do usuário transmitidas durante a autenticação de entrada do agente à sessão atual. Em outras palavras, a session_uri está vinculada ao usuário que invocou o agente inicialmente.

   A session_uri em cada URL de autorização OAuth gerada é única, e toda a URL de autorização OAuth só pode ser usada uma vez.

3. Autorize e obtenha um Access Token. A aplicação exibe a URL de autorização OAuth ao usuário para autorização. Depois que o usuário conclui o login e a autorização no DingTalk, a aplicação ou navegador envia o código de autorização OAuth ao Agent Identity por meio de redirecionamento. Ao receber o código de autorização, o Agent Identity retorna os parâmetros session_uri e state para o endereço de callback da aplicação. A aplicação verifique o parâmetro state e as informações de login do usuário atual e, em seguida, invoca a interface CompleteResourceTokenAuth do Agent Identity para continuar o processo de obtenção de um Access Token. Para obter mais informações, consulte Vinculação de sessão.

4. Reinvoque o agente para obter o Access Token. Após a aplicação invocar com êxito a interface CompleteResourceTokenAuth, o agente poderá obter o Access Token associado ao solicitante inicial do Agent Identity.

5. Acesse o recurso. O agente utiliza o Access Token obtido para acessar o recurso.

Vinculação de sessão

Se um usuário encaminhar uma URL de autorização OAuth para outra pessoa, o usuário inicial poderá obter o Access Token e os privilégios dessa pessoa. A vinculação de sessão impede isso garantindo que o usuário que obteve a URL de autorização seja o mesmo que conclui a autorização OAuth.

A implementação ocorre da seguinte forma:

1. A aplicação deve registrar um endereço de callback no Agent Identity. Isso é definido quando você configure o Workload Identity.

2. O Agent Identity não solicita imediatamente um Access Token após receber o código de autorização OAuth. Em vez disso, ele retorna a session_uri, que está vinculada ao invocador inicial, e o parâmetro state, que foi passado pela aplicação ao invocar o agente, para o endereço de callback da aplicação.

3. Adicione código de processamento no endereço de callback monitorado pela aplicação. Esse código confirma que o usuário que concluiu a autorização OAuth é o mesmo que o invocador inicial, verificando o parâmetro state e as informações de cookie na solicitação. Após a confirmação, a aplicação invoca a interface CompleteResourceTokenAuth do Agent Identity e passa a session_uri e as informações de login do usuário atual, como user_id ou JWT, para continuar a obtenção do OAuth Access Token.

4. O Agent Identity compara as informações de usuário passadas, obtidas do ID de usuário ou token de usuário, com as informações do invocador inicial na session_uri. Se houver correspondência, o Agent Identity continua a obter o Access Token usando o código de autorização OAuth e retorna uma mensagem de sucesso para a aplicação.

O exemplo Python a seguir mostra como lidar com o callback:

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

Armazenamento e renovação de tokens

O Agent Identity criptografa e armazena o OAuth Access Token e o Refresh Token obtidos (se retornados pelo provedor de credenciais) no Token Vault. Nas solicitações subsequentes, o Agent Identity retorna um Access Token válido armazenado em cache. Caso o Access Token tenha expirado, o Agent Identity usa o Refresh Token para obter um novo do provedor de credenciais. Os usuários não precisam autorizar novamente antes que o Refresh Token expire.

Se o Refresh Token tiver expirado ou se tornado inválido, o Agent Identity retornará a URL de autorização OAuth novamente, exigindo que o usuário se autentique e autorize novamente.

Uso

Após crie um provedor de credenciais OAuth no Agent Identity, use o SDK do Agent Identity para obter um OAuth Access Token. Adicione a anotação @requires_access_token antes de qualquer função que precise de um Access Token e passe os seguintes parâmetros. O SDK obtém automaticamente um Workload Access Token e o utiliza para solicitar o OAuth Access Token.

• Nome do provedor de credenciais OAuth: O nome do provedor de credenciais OAuth que você configure no Agent Identity. Para obter mais informações sobre os métodos de configuração, consulte Gerencie provedores de credenciais OAuth.

• Informações de contexto do usuário (opcional): Contém informações do usuário para uso do agente, como um ID de usuário e token de usuário (ID Token). O SDK do Agent Identity usa as informações de contexto do usuário para determinar como obter o Workload Access Token.

O exemplo Python a seguir mostra como obter um OAuth Access Token em um cenário de delegação de usuário:

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)

Referências

Usar o Agent Identity no high-code do Alibaba Cloud Model Studio