エージェントのデプロイメント

23 エージェントのデプロイメント

エージェントをデプロイすると、エージェントがホスト・アプリケーションになります。

エージェントは、プレイグラウンドまたは別のAIコンピュートにアタッチされているのと同じAIコンピュートにデプロイできます。アタッチされたAIコンピュートにエージェントに最新の変更をデプロイすると、デプロイされたエージェントはデプロイメント時のエージェントのスナップショットを表します。デプロイされたエージェントを最新バージョンに更新するには、エージェントを再デプロイする必要があります。

各エージェントには、一意のエージェント・キーに依存する安定したデプロイメントURLがあります。エージェントを複数回再デプロイすると、デプロイメントURLの背後にあるエージェントが上書きされます。

エージェントのデプロイには、次の制限があります。

エージェントは、いつでも1つのAIコンピュート・クラスタにのみデプロイできます。
同じAIコンピュート・クラスタに同じエージェントを複数回デプロイすると、以前にデプロイされたエージェントの反復が上書きされます。

エージェントをデプロイしたら、チャットURIを取得して、プログラムで問合せを発行し、エージェントの「詳細」タブからエージェントからレスポンスを取得できます。

エージェント・ページが開き、「詳細」タブが開いて強調表示されています。AIコンピュートおよびエンドポイントURLにデプロイ済が強調表示されます

エンドポイントURLは安定しており、各エージェントに関連付けられています。URLには、各エージェントに割り当てられた一意のagentIDが含まれます。つまり、エージェントをアンデプロイして再度デプロイしても、URLは同じままです。利点は、エンドポイントをコールするクライアント・コードを変更する必要がないことです。デメリットは、本番環境でエージェントを上書きできることです。

URLの構造は次のとおりです。

https://gateway.aidp.{oci-region}.oci.oraclecloud.com/agentendpoint/{agentId}/{protocol}

説明:

oci-regionは、AI Data Platformインスタンス・リージョンに対応します。
agentIdは、エージェントに関連付けられた一意のIDです。
protocolは通信プロトコルです。chatはOpenAIレスポンスAPI形式に従い、a2aはエージェント間通信プロトコルに従います。両方のプロトコルは、エージェント・エンドポイントごとに使用できます。詳細は、「A2Aエージェントのデプロイ」を参照してください。

ノート:

「Details」タブに2つのAI計算が一覧表示されます。「AIコンピュートにアタッチ」は、プレイグラウンドでエージェントをテストするために使用されます。「AIコンピュートにデプロイ済」は、デプロイされたエージェントをホストします。

「エンドポイントURL」フィールドは、エージェントのデプロイ後に移入されます。このエンドポイントURLは、本番アプリケーションからコールできます。

エージェントのデプロイ

作成して構成したエージェントをデプロイし、他のユーザーがAI Data Platformインスタンスで表示して使用できるようにします。

ホーム・ページで、デプロイするエージェントを含むフォルダに移動します。
エージェントの横にある「アクション」をクリックし、「デプロイ」をクリックします。エージェント名をクリックして、右上の「デプロイ」をクリックすることもできます。
デプロイされたエージェントにアタッチするAIコンピュートを選択します。
承認タイプとして「AIDPワークベンチ」を選択します。
セッション・データ保持ポリシーを選択します。
- 「保存期間」には、セッション・データが保持される日数を指定します。
- 「セッション・サイズ制限」には、セッションが到達できる最大サイズを指定します。
- 「スレッド数制限」には、保持されるセッション・スレッドの最大数を指定します。
「デプロイ」をクリックします

OAuth2でのエージェントのデプロイ

作成して構成したエージェントをデプロイして、OAuth2認証を使用して外部アイデンティティ・プロバイダに接続できます。

ホーム・ページで、デプロイするエージェントを含むフォルダに移動します。
エージェントの横にある「アクション」をクリックし、「デプロイ」をクリックします。エージェント名をクリックして、右上の「デプロイ」をクリックすることもできます。
デプロイされたエージェントにアタッチするAIコンピュートを選択します。
認可タイプの「OAuth2」を選択します。
オーディエンス請求を指定します。AI Data Platform Workbenchはこのフィールドに自動的に移入されますが、アイデンティティ・プロバイダからのオーディエンス要求に置換できます。
発行者要求およびJWKSを取得するURIを指定します。この情報は、アイデンティティ・プロバイダから取得されます。
セッション・データ保持ポリシーを選択します。
「デプロイ」をクリックします

エージェントのアンデプロイ

MANAGE権限を持つエージェントをアンデプロイし、使用できないようにすることができます。

ホーム・ページで、アンデプロイするエージェントが含まれているフォルダに移動します。
エージェントの横にある「アクション」をクリックし、「アンデプロイ」をクリックします。
「アンデプロイ」をクリックします。

A2Aエージェント・デプロイメント

Agent2Agent(A2A)プロトコルは、異なるフレームワークで構築されたエージェント、異なるベンダーによってホストされたエージェント、または不透明なリモート・システムとして実行されるエージェントなど、独立したAIエージェント間の通信のためのオープン・スタンダードです。

その目的は、これらのエージェントに共有インタラクション・モデルを提供して、互いの能力を発見し、サポートされている入出力形式を交渉し、タスクを委任または協力し、内部メモリー、ツールまたは実装の詳細を公開せずに情報を安全に交換できるようにすることです。詳細は、Agent2Agent (A2A)プロトコルを参照してください。

A2Aは、エージェントの相互運用性を解決することを目的としています。すべてのエージェント統合がカスタムであるのではなく、クライアントまたは別のエージェントは、共通の概念と操作セットを使用して、任意のA2A準拠のリモート・エージェントと対話できます。この仕様は、メッセージ、タスク、パーツ、アーティファクト、ストリーミング更新およびプッシュ通知を中心にしており、同期応答、長時間実行される非同期作業、ストリーミング、およびエンタープライズスタイルの認証/セキュリティ・パターンをサポートしています。

Oracle AI Data Platformでは、デプロイされたすべてのエージェントに、A2Aクライアント・アプリケーションからコールできる/A2A呼出しパスが提供されます。

エージェントカードとは?

エージェント・カードは、A2Aサーバーによって公開されるJSONメタデータ・ドキュメントです。AIDPでは、A2Aサーバーはエージェント・デプロイメントをホストするAIコンピュートです。

このカードは、エージェントのID、サービス・エンドポイント、サポートされているプロトコル/トランスポート、機能、スキル、サポートされている入出力モードおよび認証要件を記述します。クライアントはこれを使用して、エージェントが適切かどうか、およびエージェントの呼出し方法を検出します。適切に文書化されたエージェント・カードは、A2Aプロトコルの要件です。

AI Data Platform Workbenchのエージェント・カードは、「ドラフト」状態(エージェントがデプロイされていないことを意味する)または「公開済」のいずれかであり、カードがエージェントとともにデプロイされたことを意味します。

エージェント・カード処理

エージェントの開発中に、エージェントの「アクション」メニューでカードを使用できます。

次の2つのエージェント・カードにアクセスできます。

ドラフト・カードには、開発中のエージェントの現在の状態が反映されます。
公開済カードは、エージェントのデプロイ時に取得されたカードのスナップショットに対応しています。公開済カードには、デプロイ済エージェントの状態が反映されます。

エージェント・カード・フィールド

AI Data Platform Workbenchでは、現在のA2Aプロトコル・エージェント・カード・フィールドのサブセットがサポートされています(A2Aプロトコル- エージェント・カード)。

フィールド	必須	説明
`name`	○	人間が判読可能なエージェント名。例: "レシピ・エージェント"
`description`	○	エージェントの人間が読める説明で、ユーザーや他のエージェントがその目的を理解するのに役立ちます。例: 「レシピと料理をユーザーに提供するエージェント」。
`Agent Version`	○	エージェントのバージョン。例: "1.0.0"
`Documentation URL`	×	エージェントに関する追加のドキュメントを提供するURL。
`Provider - Organization`	×	エージェントのサービス・プロバイダ。
`Provider - URL`	×	サービス・プロバイダのURL。
`Capabilities`	○	エージェントがサポートするA2A機能セット。 `streaming`のみ構成できます(True/False)
`Skills`	○	スキルは、エージェントの能力を表します。これは主に説明的な概念ですが、エージェントが成功する可能性が高い、より焦点を絞った一連の動作を表します。スキルはAgentSkillの配列を表します。

各AgentSkillは、エージェントの機能をドキュメント化する複数のフィールドで構成されています。エージェント・カードでのエージェント・スキルの定義は最も時間がかかる操作であり、反復プロセスです。スキルは、デプロイ前にエージェント・カードのドラフトで(エージェント・カードの残りとともに)編集できます。

ノート:

inputModes、outputModesおよび securityRequirementsは、AI Data Platform Workbenchによって提供されており、変更できません。

フィールド	必須	説明
`Skill ID`	○	エージェントのスキルの一意の識別子。
`Skill Name`	○	スキルの判読可能な名前。
`Description`	○	スキルの詳しい説明。
`Tags`	○	スキルの機能を記述する一連のキーワード。
`Examples`	×	このスキルが処理できるプロンプトまたはシナリオの例。

エージェント・デプロイメント・エンドポイントA2Aパス

/a2aパスは、デプロイされたエージェントのURLに/chatに加えて公開されます。

たとえば、エージェントはこれらのパスを外部クライアントに公開します。

https://gateway.aidp.{oci-region}.oci.oraclecloud.com/agentendpoint/{agentId}/chat
https://gateway.aidp.{oci-region}.oci.oraclecloud.com/agentendpoint/{agentId}/a2a

両方のパス(/chat、/a2a)を別々のクライアントで使用できます。

A2Aのセッション変数

セッション変数の値は、メッセージmetadataフィールドでA2Aエージェントに渡すことができます。次のJSONスニペットは、userName、geoLocationおよびosの3つのセッション変数を使用してa2aエージェントに発行されたユーザー・メッセージのペイロードを示しています。

{
  "jsonrpc": "2.0",
  "method": "message/send",
  "params": {
    "contextId": "session_12345",
    "taskId": "task_67890",
    "message": {
      "role": "user",
      "parts": [
        {
          "text": "What is the current status of my order?",
        }
      ],
      "metadata": {
        "sessionvariables.userName": "George",
        "sessionvariables.geoLocation": “Dallas, TX”,
        "sessionvariables.os": "mobile_ios"
      }
    }
  },
  "id": "rpc-99821"
}

例: OCI CLIを使用したA2Aエージェントの起動(非ストリーミング)

oci raw-request \
  --http-method POST \
  --auth security_token \
  --request-body '{
    "id": "<your-request-id>",
    "jsonrpc": "2.0",
    "method": "message/send",
    "params": {
      "configuration": {
        "acceptedOutputModes": [
          "text/plain",
          "text"
        ]
      },
      "message": {
        "contextId": "<your-context-id>",
        "kind": "message",
        "messageId": "<your-message-id>",
        "parts": [
          {
            "kind": "text",
            "text": "What is the capital of India?"
          }
        ],
        "role": "user"
      }
    }
  }' \
  --request-headers '{
    "x-session-id": "<your-session-id>",
    "dh-user-principal": "<your-user-principal>"
  }' \
  --target-uri " <your-a2a-agent-endpoint-url>"

例: OCI CLIを使用したA2Aエージェントの起動(ストリーミング)

oci raw-request \
  --http-method POST \
  --auth security_token \
  --request-body '{
    "id": "<your-request-id>",
    "jsonrpc": "2.0",
    "method": "message/stream",
    "params": {
      "configuration": {
        "acceptedOutputModes": [
          "text/plain",
          "text"
        ]
      },
      "message": {
        "contextId": "<your-context-id>",
        "kind": "message",
        "messageId": " <your-message-id>",
        "parts": [
          {
            "kind": "text",
            "text": "What is the capital of India?"
          }
        ],
        "role": "user"
      }
    }
  }' \
  --request-headers '{
    "x-session-id": "<your-session-id>",
    "dh-user-principal": "<user-principal>"
  }' \
  --target-uri "<your-a2a-agent-endpoint-url>"

例: A2AクライアントSDK

import asyncio
import json
import logging
import typing
from collections.abc import Iterator
import uuid
import httpx
import oci
from a2a.client import A2AClient, ClientFactory
from a2a.types import (
    AgentCard,
    Message,
    Part,
    Role,
    TextPart,
    SendMessageRequest,
    MessageSendParams,
    MessageSendConfiguration,
    Task, SendMessageSuccessResponse, SendStreamingMessageRequest,
)

class OCIAuth(httpx.Auth):
    """httpx auth implementation using OCI signer via requests auth adapter."""

    def __init__(self, signer: oci.signer.AbstractBaseSigner):
        self._requests_auth = _OCIRequestsAuth(signer)

    def auth_flow(self, request: httpx.Request) -> Iterator[httpx.Request]:
        req = RequestsRequest(
            method=request.method,
            url=str(request.url),
            headers=dict(request.headers),
            data=request.content,
        )
        prepared: RequestsPreparedRequest = req.prepare()
        prepared = self._requests_auth(prepared)
        request.headers.update(dict(prepared.headers))
        yield request



def getOCIAuth():
    conf = oci.config.from_file(profile_name="DEFAULT")
    token_file = conf['security_token_file']
    token = None
    with open(token_file, 'r') as f:
        token = f.read()
    private_key = oci.signer.load_private_key_from_file(conf['key_file'])
    signer = oci.auth.signers.SecurityTokenSigner(token, private_key)
    auth = OCIAuth(signer=signer)
    return auth


async def _call_agent_with_a2a(agent_url: str, query: str, context_id: str,auth:OCIAuth) -> str:
    """Call an agent using the A2A protocol."""
    try:
        # Initialize OCI signer
        #headers = {"dh-user-principal": "dh-user"}
        headers = {"Accept": "*/*",
                   "dh-user-principal": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}

        async with httpx.AsyncClient(timeout=60.0, auth=auth,headers=headers) as hc:
            agent_card = await _get_agent_card(agent_url,auth)
            print(f"Agent card is {agent_card}")

            client =  A2AClient(httpx_client=hc, agent_card=agent_card)
            # Create message
            message = Message(
                message_id=str(uuid.uuid4()),
                context_id=context_id,
                role=Role.user,
                parts=[Part(root=TextPart(text=query))],
                metadata={"sessionvariables.cred.mcp.weatherReportMCP.bearer": "valid-123"}
            )
            request = SendMessageRequest(
                id=str(uuid.uuid4()),  # Add the required id field
                params=MessageSendParams(
                    message=message,
                    configuration=MessageSendConfiguration(acceptedOutputModes=["text/plain", "text"]),
                ),
            )
            #json_string = json.dumps(message, indent=4)
            print(f"Send request : {request}")

            response = await client.send_message(request)

            logging.info("Received response from A2A server: %s", response.root.result)
            # Extract response
            result = response.root.result
            # Handle different response types


            if isinstance(result, Task):
                # Task response
                if result.artifacts:
                    # Extract text from artifacts
                    texts = []
                    for artifact in result.artifacts:
                        for part in artifact.parts:
                            if hasattr(part, "root") and hasattr(part.root, "text"):
                                texts.append(part.root.text)
                    return "\n".join(texts) if texts else "Task completed with no text response"
                elif result.status and result.status.message:
                    logging.info(f"Received Task status {result.status.state} from A2A server and status message is {result.status.message}", result.status.message)
                    if  result.status.state== "failed":
                        print("Failure observed in Task invocation")
                        for m_part in result.status.message.parts:
                           print(f"Error message  { m_part.root.text}")

                    return get_message_text(result.status.message)
                else:
                    return f"Task {result.id} status: {result.status.state if result.status else 'unknown'}"

            elif isinstance(result, Message):
                return get_message_text(result)
            else:
                logging.warning(f"Unexpected response type: {type(result)}")
                return "Received response but unable to extract text"
    except Exception as ex:
        logging.error(f"Error calling agent at {agent_url}: {ex}", exc_info=True)
        return f"Error communicating with agent: {str(ex)}"


async def _call_agent_with_a2a_with_stream(agent_url: str, query: str, context_id: str, auth: OCIAuth) -> str:
    """Call an agent using the A2A protocol with streaming (SSE) and return the final artifact text."""
    try:
        async with httpx.AsyncClient(timeout=60.0, auth=auth) as hc:
            agent_card = await _get_agent_card(agent_url)
            if not agent_card:
                return "No Agent Card Found"
            print(f"Agent card is {agent_card}")

            client = A2AClient(httpx_client=hc, agent_card=agent_card)
            message = Message(
                message_id=str(uuid.uuid4()),
                context_id=context_id,
                role=Role.user,
                parts=[Part(root=TextPart(text=query))],
            )
            request = SendStreamingMessageRequest(
                id=str(uuid.uuid4()),
                params=MessageSendParams(
                    message=message,
                    configuration=MessageSendConfiguration(acceptedOutputModes=["text/plain", "text"]),
                ),
            )
            print("Invoking Remote Agent request (beautified JSON):")
            print(json.dumps(request.model_dump(), indent=2, ensure_ascii=False))
            # Expected event types:
            # - TaskStatusUpdateEvent (working/in-progress)
            # - TaskArtifactUpdateEvent (contains Artifact.parts[].root.text) -> final output
            final_artifact_text_parts: list[str] = []

            async for event in client.send_message_streaming(request):
                # Print each SSE event as-is (SDK object)
                print(f"[A2A stream event] {event}")

                try:
                    result = getattr(event.root, "result", None)
                    if not result:
                        continue

                    # TaskArtifactUpdateEvent and TaskStatusUpdateEvent are SDK types; to avoid tight coupling,
                    # extract by attribute presence.
                    artifact = getattr(result, "artifact", None)
                    if artifact and getattr(artifact, "parts", None):
                        for part in artifact.parts:
                            root = getattr(part, "root", None)
                            txt = getattr(root, "text", None)
                            if txt:
                                final_artifact_text_parts.append(txt)
                except Exception:
                    # Keep streaming even if an event can't be parsed
                    continue

            return "\n".join([t for t in final_artifact_text_parts if t]).strip() or "Stream completed (no artifact text)."
    except Exception as ex:
        logging.error(f"Error calling agent at {agent_url}: {ex}", exc_info=True)
        return f"Error communicating with agent: {str(ex)}"

ドラフト・エージェント・カードの編集

まだデプロイされていないエージェントのエージェント・カードを編集できます。

エージェントにナビゲートします。
「アクション」をクリックし、「エージェント・カードの表示」および「エージェント・カードのドラフト」をクリックします。
「Edit」をクリックします。
右上の「フォーム」または「JSON」をクリックすると、ビューを切り替えることができます。JSONビューはより完全ですが、読取り専用です。フィールドは、フォーム・ビューでのみ編集できます。
必要に応じてフィールドを変更します。
「スキルの追加」をクリックして、A2Aクライアントに公開するスキルを追加します。
「保存」をクリックします

公開済エージェント・カードの編集

エージェントをアンデプロイまたは再デプロイしなくても、公開されたエージェント・カードを変更できます。

公開済カードは、デプロイメント時のドラフト・カードのスナップショットに対応しています。

ノート:

公開されたカードに加えた変更は、A2Aクライアントからアクセス可能なagent-card.jsonファイルにすぐに反映されます。

エージェントにナビゲートします。
「アクション」をクリックし、「エージェント・カードの表示」および「公開済エージェント・カード」をクリックします。
右上の「フォーム」または「JSON」をクリックすると、ビューを切り替えることができます。JSONビューはより完全ですが、読取り専用です。フィールドは、フォーム・ビューでのみ編集できます。
「Edit」をクリックします。
必要に応じてフィールドを変更します。
「スキルの追加」をクリックして、A2Aクライアントに公開するスキルを追加します。
「変更の公開」をクリックします。

公開されたエージェント・カードは公開されません。ユーザーを認証し、agent-card.jsonの内容を検査するための適切な権限(READ)を持っている必要があります。他のエージェントは、標準パスでHTTP GETリクエストを介してエージェントの機能とメタデータを検出できます。

https://gateway.aidp.{oci-region}.oci.oraclecloud.com/agentendpoint/{agentId}/a2a/agent-card.json