Skip to content
For Self-Managed AI Platform users running v11.1, see the on-premise platform documentation
エージェント型AI > エージェントのワークフローを開発 > リクエストヘッダーへのアクセス

リクエストヘッダーへのアクセス

エージェントがデプロイされると、認証、追跡、またはカスタムメタデータのために、HTTPリクエストヘッダーにアクセスする必要があるかもしれません。 DataRobot はchat()関数の**kwargsパラメーターを通して、エージェントコードからヘッダーを利用できるようにします。

X-Untrusted-*ヘッダーの抽出

X-Untrusted-*プレフィックスを持つヘッダーは、元のリクエストから渡されます。 They are available in the kwargs dictionary, located in agent/agentic_workflow/custom.py:

agent/agentic_workflow/custom.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Iterator, Union
from openai.types.chat import CompletionCreateParams
from openai.types.chat.completion_create_params import (
    CompletionCreateParamsNonStreaming,
    CompletionCreateParamsStreaming,
)

def chat(
    completion_create_params: CompletionCreateParams
    | CompletionCreateParamsNonStreaming
    | CompletionCreateParamsStreaming,
    load_model_result: tuple[ThreadPoolExecutor, asyncio.AbstractEventLoop],
    **kwargs: Any,
) -> Union[CustomModelChatResponse, Iterator[CustomModelStreamingResponse]]:
    # Extract all headers from kwargs
    headers = kwargs.get("headers", {})

    # Access specific X-Untrusted-* headers
    authorization_header = headers.get("X-Untrusted-Authorization")
    custom_header = headers.get("X-Untrusted-Custom-Metadata")

    # Use headers in your agent logic
    if authorization_header:
        # Pass to downstream services, tools, etc.
        # Do not log raw authorization tokens; log only non-sensitive metadata.
        request_id = headers.get("X-Untrusted-Request-ID")
        if request_id:
            print(f"Processing request {request_id} with authorization header present")
        else:
            print("Processing request with authorization header present")

    # Continue with agent logic...

一般的なユースケース

リクエストヘッダーは、エージェントのワークフローでさまざまな目的に使用できます。 以下の例は、ヘッダー情報を抽出して使用するための一般的なパターンを示します。

外部サービスに認証を渡す

リクエストヘッダーから認証トークンを抽出し、外部サービスに渡します。 これにより、受信リクエストから下流のツールおよびサービスに資格情報を転送できます。

agent/agentic_workflow/custom.py
headers = kwargs.get("headers", {})
auth_token = headers.get("X-Untrusted-Authorization")

# Use the token to authenticate with external APIs
tool_client = MyTool(auth_token=auth_token)

リクエストメタデータの追跡

ヘッダーを使用して、デバッグや分析のためにリクエストID、ユーザーID、その他のメタデータを追跡します。 これにより、エージェントワークフローを通じてリクエストを追跡できます。

agent/agentic_workflow/custom.py
headers = kwargs.get("headers", {})
request_id = headers.get("X-Untrusted-Request-ID")
user_id = headers.get("X-Untrusted-User-ID")

# Log metadata for debugging or analytics
logging.info(f"Processing request {request_id} for user {user_id}")

条件付きのエージェントの動作

ヘッダーからのリクエストコンテキストに基づいて、エージェントの動作を調整します。 これにより、さまざまなリージョン、ユーザー、またはデプロイコンテキストに対してエージェントの動作をカスタマイズできます。

agent/agentic_workflow/custom.py
headers = kwargs.get("headers", {})
region = headers.get("X-Untrusted-Region")

# Adjust agent behavior based on request context
if region == "EU":
    agent = MyAgent(enable_gdpr_mode=True)

重要な注意事項

エージェントコードでリクエストヘッダーを扱う際には、以下の点に注意してください。

  • X-Untrusted-*プレフィックスを持つヘッダーのみが、エージェントコードに渡されます。
  • ディクショナリからヘッダーにアクセスする場合、英大文字と英小文字は区別されます。
  • 存在しない可能性のあるヘッダーにアクセスする場合、常にデフォルト値を指定するか、「なし」をチェックします。
  • ヘッダーは、ストリーミングと非ストリーミングの両方のチャット回答で使用できます。

デバッグのためリクエストヘッダーのログを取る

使用可能なすべてのヘッダーをログに記録して、エージェントに渡される内容を調べます。

agent/agentic_workflow/custom.py
import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
from typing import Any

def chat(completion_create_params, load_model_result: tuple[ThreadPoolExecutor, asyncio.AbstractEventLoop], **kwargs: Any):
    headers = kwargs.get("headers", {})

    # Log all headers to inspect what's available
    if headers:
        logging.warning(f"All headers: {dict(headers)}")

    # Continue with agent logic...