23 代理部署

部署代理会将您的代理转换为托管应用程序。

您可以将代理部署到附加到其操场或不同 AI 计算的同一 AI 计算中。将代理的最新更改部署到附加的 AI 计算时，部署的代理表示部署时代理的快照。要将部署的代理更新到最新版本，需要重新部署代理。

每个代理都有一个依赖于唯一代理键的稳定部署 URL。多次重新部署代理会覆盖部署 URL 后面的代理。

部署代理具有以下限制：

• 在任意给定时间，一个代理只能部署到一个 AI 计算集群。
• 将同一代理多次部署到同一 AI 计算集群会覆盖以前部署的代理迭代。

部署座席后，您可以检索聊天 URI，以通过编程方式发出查询并从座席的详细信息选项卡检索座席的响应。

端点 URL 是稳定的，它与每个代理绑定。URL 包括分配给每个代理的唯一 agentID。换句话说，如果您取消部署代理并重新部署该代理，则 URL 保持不变。优点是您不必修改调用端点的客户端代码，缺点是您可以覆盖生产环境中的代理。

URL 具有以下结构：

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

其中：

• oci-region 对应于 AI 数据平台实例区域；
• agentId 是与代理关联的唯一 ID
• protocol 是通信协议：chat 遵循 OpenAI Responses API 格式，a2a 遵循代理到代理通信协议。这两种协议都可用于每个代理端点。有关更多信息，请参见 A2A Agent Deployment 。

注意：

"Details"（详细信息）选项卡中列出了两个 AI 计算。附加到 AI 计算用于在操场中测试代理。Deployed to AI Compute 托管部署的代理。

部署代理后，将填充端点 URL 字段。您可以从生产应用程序调用此端点 URL。

部署代理

您可以部署已创建和配置的代理，以便其他用户能够在 AI 数据平台实例中查看和使用它。

1. 在主页上，导航到包含要部署的代理的文件夹。
2. 在代理旁边，单击 操作，然后单击部署。您还可以单击代理名称，然后单击右上角的部署。
   
   

   

   
   
3. 选择要附加到已部署代理的 AI 计算。
4. 为授权类型选择 AIDP 工作台。
5. 选择会话数据保留策略。
   • 对于保留期，提供会话数据的保留天数。
   • 对于会话大小限制，提供会话可以达到的最大大小。
   • 对于线程计数限制，提供保留的最大会话线程数。
6. 单击部署。

使用 OAuth2 部署代理

您可以部署已创建并配置为使用 OAuth2 验证的代理以连接到外部身份提供者。

1. 在主页上，导航到包含要部署的代理的文件夹。
2. 在代理旁边，单击 操作，然后单击部署。您还可以单击代理名称，然后单击右上角的部署。
   
   

   

   
   
3. 选择要附加到已部署代理的 AI 计算。
4. 为授权类型选择 OAuth2 。
5. 提供受众索赔。AI Data Platform Workbench 会自动填充此字段，但您可以将其替换为来自身份提供商的受众声明。
6. 提供 Issuer 索赔和 URI 以检索 JWKS 。此信息源自您的身份提供者。
7. 选择会话数据保留策略。
8. 单击部署。

取消部署代理

您可以选择取消部署您具有 MANAGE 权限的代理，使其无法使用。

1. 在主页上，导航到包含要取消部署的代理的文件夹。
2. 在代理旁边，单击 操作，然后单击取消部署。
   
   

   

   
   
3. 单击取消部署。

A2A 代理部署

Agent2Agent(A2A) 协议是独立 AI 代理之间通信的开放标准，包括使用不同框架构建的代理，由不同的供应商托管，或作为不透明的远程系统运行。

其目的是为这些代理提供共享交互模型，以便他们能够发现彼此的能力，协商支持的输入/输出格式，委托或协作处理任务，并安全地交换信息，而不会暴露内部内存、工具或实施详细信息。有关更多信息，请参见 Agent2Agent (A2A) Protocol 。

A2A 旨在解决代理互操作性：客户机或其他代理可以使用一组通用的概念和操作与任何符合 A2A 标准的远程代理进行交互，而不是定制每个代理集成。该规范以消息、任务、部件、构件、流更新和推送通知为中心；它支持同步回复、长时间运行的异步工作、流处理和企业式验证/安全模式。

在 Oracle AI Data Platform 中，所有部署的代理都有一个 /A2A 调用路径，A2A 客户机应用程序可以调用该路径。

什么是代理卡？

代理卡是由 A2A 服务器发布的 JSON 元数据文档。在 AIDP 中，A2A 服务器是托管代理部署的 AI 计算。

卡描述代理的身份、服务端点、支持的协议/传输、功能、技能、支持的输入/输出模式以及身份验证要求；客户机使用它来发现代理是否合适以及如何调用代理。正确记录的代理卡是 A2A 协议的要求。

AI 数据平台工作台中的座席卡处于“草稿”状态，这意味着座席尚未部署，或者已发布，这意味着该卡已与座席一起部署。

代理卡操作

在代理的开发过程中，该卡在代理的“操作”菜单中可用。

可访问两个代理卡：

• 草稿卡反映了代理在开发中的当前状态。
• 发布的卡对应于部署代理时获取的卡的快照。发布的卡反映了已部署代理的状态。

代理卡字段

AI Data Platform Workbench 支持当前 A2A 协议代理卡字段的子集，可在此处获取： A2A Protocol - Agent Card 。

域	必需	说明
name	是	代理的人类可读名称。示例：“配方代理”
description	是	代理人可读的描述，帮助用户和其他代理人了解其目的。示例：“帮助用户制作食谱和烹饪的代理”。
Agent Version	是	代理的版本。示例："1.0.0"
Documentation URL	无	提供有关代理的其他文档的 URL。
Provider - Organization	无	代理的服务提供方。
Provider - URL	无	服务提供者的 URL。
Capabilities	是	代理支持的 A2A 功能集。 只能配置 streaming (True/False)
Skills	是	技能表示座席的能力。它基本上是一个描述性的概念，但代表了一组更集中的行为，代理可能会成功。技能代表 AgentSkill 的数组。

每个 AgentSkill 由多个字段组成，用于记录 Agent 的功能。在代理卡中定义代理技能是最耗时的操作，是一个迭代过程。在部署之前，可以在草稿座席卡中编辑技能（以及座席卡的其余部分）。

注意：

inputModes、outputModes 和 securityRequirements 由 AI Data Platform Workbench 提供，无法修改。

域	必需	说明
Skill ID	是	代理技能的唯一标识符。
Skill Name	是	技能的人类可读名称。
Description	是	技能的详细说明。
Tags	是	描述技能功能的一组关键字。
Examples	无	此技能可以处理的提示或方案示例。

代理部署端点 A2A 路径

除 /chat 之外，/a2a 路径在已部署代理的 URL 中公开。

例如，代理将向外部客户机公开以下路径：

• 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 片段显示了向 a2a 代理发出的具有三个会话变量的用户消息的有效负载：userName、geoLocation 和 os：

{
  "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)}"

编辑草稿代理卡

您可以为尚未部署的代理编辑代理卡。

1. 导航到您的代理。
2. 单击操作，然后单击查看代理卡和草稿代理卡。
   
   

   

   
   
3. 单击编辑。
   
   

   

   
   
4. 您可以通过单击右上角的表单或 JSON 来切换视图。JSON 视图更完整，但为只读。您只能编辑“表单”视图中的字段。
   
   

   

   
   
5. 根据需要修改字段。
6. 单击添加技能以添加要向 A2A 客户机公开的技能。
   
   

   

   
   
7. 单击保存。

编辑已发布的座席卡

您可以修改已发布的代理卡，而无需取消部署或重新部署代理。

发布的卡对应于部署时草稿卡的快照。

注意：

对已发布卡所做的更改会立即反映在 A2A 客户机可访问的 agent-card.json 文件中。

1. 导航到您的代理。
2. 单击操作，然后单击查看代理卡和已发布的代理卡。
   
   

   

   
   
3. 您可以通过单击右上角的表单或 JSON 来切换视图。JSON 视图更完整，但为只读。您只能编辑“表单”视图中的字段。
4. 单击编辑。
   
   

   

   
   
5. 根据需要修改字段。
6. 单击添加技能以添加要向 A2A 客户机公开的技能。
   
   

   

   
   
7. 单击发布更改。

已发布的座席卡未公开可用。需要对用户进行身份验证，并具有检查 agent-card.json 内容的正确权限 (READ)。其他代理可以通过标准路径上的 HTTP GET 请求来搜索代理功能和元数据：

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