エージェント認証¶

import os
import requests
from crewai.tools import BaseTool

class ExternalAPITool(BaseTool):
    def run(self, query: str) -> str:
        api_key = os.getenv("EXTERNAL_API_KEY")
        headers = {"Authorization": f"Bearer {api_key}"}
        response = requests.get(
            "https://api.external-service.com/data",
            headers=headers,
            params={"query": query}
        )
        return response.json() 

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

# In your chat() function
initialize_authorization_context(completion_create_params)

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

from datarobot_genai.core.chat.client import ToolClient

tool_client = ToolClient()
result = tool_client.call(
    deployment_id="your_deployment_id",
    payload={"query": "your_query"}
) 

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

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

from agent_generic_base.custom_model.agent import MyAgent

agent = MyAgent(verbose=True) 

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

from datarobot_genai.core.cli import AgentEnvironment

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

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

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