エージェント認証¶

AIエージェントは、タスクを完了するために外部リソースへの認証を必要とすることがよくあります。 たとえば、デプロイされたエージェントは、データを取得したり操作を実行したりするために、外部API、データベース、またはクラウドサービスにアクセスする必要がある場合があります。

This documentation provides comprehensive guidance for implementing authentication in DataRobot Agent Templates, covering API tokens, MCP server authentication, authorization context, OAuth 2.0, and security best practices.

認証方法¶

このセクションでは、フレームワークで利用可能なさまざまな認証方式の概要を説明します。

APIトークン認証¶

APIトークン認証は、DataRobotのサービスや外部APIへの認証において最も一般的な方法です。 ヘッダーまたは環境変数で渡されるベアラートークンを使用します。

DataRobot APIの認証¶

環境変数またはプログラムを使用してAPIトークンを設定します。 DATAROBOT_API_TOKENは、アカウント設定の「APIのキーとツール」セクションにあります。

DATAROBOT_API_TOKEN=<your_api_key>
DATAROBOT_ENDPOINT=https://app.datarobot.com 

from agent import MyAgent

agent = MyAgent(
    api_key="your_api_key",
    api_base="https://app.datarobot.com"
) 

外部APIの認証¶

外部APIへの認証が必要なカスタムツールを作成する場合は、ツールのrun()メソッド内で環境変数またはランタイムパラメーターからAPIキーを取得します。 これにより、資格情報がソースコードに含まれず、エージェントテンプレート全体で使用されているのと同じセキュリティパターンに従います。

ローカルで開発する場合は、環境変数を使用します。 デプロイされたエージェントの場合は、model-metadata.yamlでランタイムパラメーターを定義し、datarobot_drumからRuntimeParameters.get()を使って取得します。 以下の例に示すように、まずos.environ.get()を試し（ローカル開発の場合）、その後、RuntimeParameters.get()にフォールバックします（デプロイされたエージェントの場合）。

import os
import requests
from crewai.tools import BaseTool
from datarobot_drum import RuntimeParameters

class ExternalAPITool(BaseTool):
    def run(self, query: str) -> str:
        api_key = os.environ.get("EXTERNAL_API_KEY")
        if not api_key:
            api_key = RuntimeParameters.get("EXTERNAL_API_KEY")["apiToken"]

        headers = {"Authorization": f"Bearer {api_key}"}
        response = requests.get(
            "https://api.external-service.com/data",
            headers=headers,
            params={"query": query}
        )
        return response.json() 

runtimeParameterDefinitions:
  - fieldName: EXTERNAL_API_KEY
    type: credential
    credentialType: api_token
    description: API key for external service authentication 

MCP server authentication¶

When using MCP (Model Context Protocol) servers to provide tools to your agents, authentication is handled automatically using the DATAROBOT_API_TOKEN environment variable. The MCP client in the agent templates automatically uses this token for authenticated requests to the MCP server.

Configuration:

DATAROBOT_API_TOKEN=<your_api_key>
DATAROBOT_ENDPOINT=https://app.datarobot.com 

The MCP client will automatically use DATAROBOT_API_TOKEN when making requests to the MCP server. No additional configuration is required beyond setting these environment variables.

認可コンテキスト¶

フレームワークは、認証情報を下流のツールやサービスに伝えるための認可コンテキストシステムを提供します。 これにより、手動で設定することなく、エージェントツール間でトークンと資格情報を自動的に受け渡すことができます。

datarobot-genaiパッケージのresolve_authorization_context()を使用し、エージェントのchat()関数で認可コンテキストを初期化します。 この関数は認可コンテキストディクショナリを返します。これは、completion_create_params["authorization_context"]に割り当てる必要があります。 その後、ツールはdatarobot SDKのget_authorization_context()を使用してコンテキストを取得できます。

from datarobot_genai.core.chat import resolve_authorization_context
from datarobot.models.genai.agent.auth import get_authorization_context

def chat(completion_create_params, load_model_result, **kwargs):
    # Initialize the authorization context for downstream agents and tools
    completion_create_params["authorization_context"] = resolve_authorization_context(
        completion_create_params, **kwargs
    )
    # ... rest of chat function

# In your tools
auth_context = get_authorization_context()
access_token = auth_context.get("access_token") 

OAuth 2.0認証¶

OAuth 2.0をサポートする外部サービスの場合は、ツールにOAuthフローを実装します。 サービスの開発者コンソールでOAuthアプリケーションを作成し、環境変数を設定します。

OAUTH_CLIENT_ID=<your_client_id>
OAUTH_CLIENT_SECRET=<your_client_secret>
OAUTH_REDIRECT_URI=<your_redirect_uri>
OAUTH_SCOPE=<required_scopes> 

次に、OAuthの実装でそれらの環境変数を使用します。 この例は、CrewAIツール向けの完全なOAuth 2.0認可コードフローを示しています。 このツールはcrewai.tools.BaseTool（CrewAIエージェントテンプレートのツールの標準基本クラス）を継承し、インメモリーキャッシュを使用して独自のトークンライフサイクルを管理し、環境変数から資格情報を取得するリポジトリのパターンに従います。

import os
import time
import requests
from urllib.parse import urlencode
from crewai.tools import BaseTool

class ExampleToolWithOAuth(BaseTool):
    def __init__(self):
        super().__init__()
        self.client_id = os.getenv("OAUTH_CLIENT_ID")
        self.client_secret = os.getenv("OAUTH_CLIENT_SECRET")
        self.redirect_uri = os.getenv("OAUTH_REDIRECT_URI")
        self._access_token = None
        self._token_expires_at = None

    def get_authorization_url(self) -> str:
        params = {
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "response_type": "code",
            "scope": "read:data"
        }
        return f"https://oauth.provider.com/authorize?{urlencode(params)}"

    def exchange_code_for_token(self, code: str) -> dict:
        response = requests.post(
            "https://oauth.provider.com/token",
            data={
                "grant_type": "authorization_code",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "code": code,
                "redirect_uri": self.redirect_uri
            }
        )
        token_data = response.json()
        self._access_token = token_data.get("access_token")
        expires_in = token_data.get("expires_in", 3600)
        self._token_expires_at = time.time() + expires_in
        return token_data

    def get_cached_access_token(self) -> str:
        if self._access_token and self._token_expires_at and time.time() < self._token_expires_at:
            return self._access_token
        # Token expired or not set - refresh or re-authenticate
        # In production, implement token refresh logic here
        raise ValueError("Access token expired. Re-authenticate to get a new token.")

    def run(self, query: str) -> str:
        access_token = self.get_cached_access_token()
        response = requests.get(
            "https://api.provider.com/data",
            headers={"Authorization": f"Bearer {access_token}"},
            params={"query": query}
        )
        return response.json() 

セキュリティのベストプラクティス¶

本番環境で認証を処理する場合、セキュリティのベストプラクティスに従うことが不可欠です。 以下のガイドラインに従ってください。

• APIトークン認証
• トークンは環境変数に保管します。ソースコードにシークレットを直接記述することは絶対に避けてください。
• 本番環境ではセキュアなトークン保管ソリューションを利用します。
• トークンのローテーションを実施し、トークンの有効期限ポリシーを適用します。
• 認可コンテキストは、検証してからツールで使用します。
• 最小権限の原則に従います。
• 監査目的で認証イベントをログに記録します。
• OAuth 2.0
• すべてのOAuth通信にHTTPSを使用します。
• CSRF攻撃を防ぐためにstateパラメーターを検証します。
• リフレッシュトークンをセキュアに保管します。
• トークンの有効期限とリフレッシュのロジックを確実に処理します。
• 認可コンテキストは、検証してから使用します。
• セキュリティ全般
• 開発と運用に別々の環境を使用します。
• 強固なシークレット管理プラクティスを実施します。
• コンテナセキュリティのベストプラクティスに従います。
• 定期的なセキュリティ監査を実施し、更新プログラムを適用します。

認証に関する問題のトラブルシューティング¶

このセクションでは、エージェントの開発またはデプロイ時に発生する一般的な認証の問題を診断し、解決する方法について説明します。

よくある問題¶

以下のセクションでは、一般的な認証エラーとその解決方法について説明します。

APIトークンがない¶

このエラーは、DataRobot APIトークンが設定されていない場合に発生します。

問題：エラー：DataRobot APIトークンがありません。 DATAROBOT_API_TOKEN環境変数を設定してください

解決方法：DataRobot APIキーを使用するようにDATAROBOT_API_TOKEN環境変数を設定します。

無効なエンドポイント¶

このエラーは、DataRobotのエンドポイントがないか、正しく設定されていない場合に発生します。

問題：エラー：DataRobotのエンドポイントがありません。 DATAROBOT_ENDPOINT環境変数を設定してください

解決方法：DATAROBOT_ENDPOINT環境変数を正しく設定します。

認可コンテキストが設定されていない¶

このエラーは、ツールが初期化前の認可コンテキストにアクセスしようとすると発生します。

問題：エラー：ツールで認可コンテキストを使用できません

解決方法：resolve_authorization_context()がエージェントのchat()関数で呼び出され、その結果がcompletion_create_params["authorization_context"]に割り当てられるようにします。

OAuthトークンの有効期限切れ¶

このエラーは、OAuthアクセストークンの有効期限が切れており、リフレッシュする必要がある場合に発生します。

問題：エラー：401 Unauthorized

解決方法：トークンリフレッシュロジックを実装するか、再認証します。

MCP server authentication failure¶

This error occurs when the MCP server cannot authenticate the agent's requests.

Issue: Error: 401 Unauthorized or MCP connection failures

Solutions:

1. Verify API token: Ensure DATAROBOT_API_TOKEN is set correctly in your environment
2. Check token permissions: Verify the token has necessary permissions for MCP server access
3. Verify MCP server endpoint: Check that the MCP server endpoint is correct
4. For local development: Ensure the MCP server is running and accessible
5. Review MCP configuration: See MCP tools troubleshooting for more details

デバッグのヒント¶

認証の問題をデバッグするのに役立つテクニックを紹介します。

冗長ログを有効にする¶

認証操作に関する詳細情報を取得するには、詳細ログを有効にします。 そのためには、custom.pyでMyAgentがインスタンス化されている場所を探し、verbose=Trueを設定します。

from agent import MyAgent

# ...

agent = MyAgent(verbose=True, **completion_create_params) 

環境変数のチェック¶

必要な環境変数が正しく設定されていることを確認します。

import os
print(f"API Token: {os.getenv('DATAROBOT_API_TOKEN')}")
print(f"Endpoint: {os.getenv('DATAROBOT_ENDPOINT')}") 

認証のテスト¶

datarobot_genai.core.cliのAgentEnvironmentクラスを使用して認証をテストします。

from datarobot_genai.core.cli import AgentEnvironment

try:
    env = AgentEnvironment()
    print("Authentication successful")
except ValueError as e:
    print(f"Authentication failed: {e}") 

認可コンテキストの検証¶

datarobot.models.genai.agent.authのget_authorization_context()を使用して、認可コンテキストを確認します。

from datarobot.models.genai.agent.auth import get_authorization_context

auth_context = get_authorization_context()
print(f"Authorization context: {auth_context}")