Contrato de ambiente de execução da Gemini Enterprise Agent Platform

O Agent Runtime foi projetado para ser independente do framework do aplicativo. Se você optar por implantar o agente usando um contêiner personalizado ou um Dockerfile, o contêiner precisará aderir ao contrato de ambiente de execução para exibir consultas.

Para mais informações sobre os métodos de implantação, consulte Implantar um agente.

Restrições e requisitos

Para implantar um contêiner personalizado no Agent Runtime, ele precisa detectar solicitações HTTP em 0.0.0.0 na porta 8080.

Endpoints (opcional)

Seu contêiner pode expor qualquer endpoint HTTP personalizado. É possível invocar esses endpoints personalizados enviando solicitações para a API do agente implantado. Para mais informações, consulte Usar agentes implantados pela API.

Embora esses endpoints sejam opcionais no nível da API, a implementação deles permite os seguintes recursos de integração:

• Suporte ao SDK do Python: a implementação de /api/reasoning_engine e /api/stream_reasoning_engine permite que você use o agente implantado com o SDK do Python da Gemini Enterprise Agent Platform.
• Suporte ao Playground: a implementação de /api/stream_reasoning_engine é necessária se você quiser interagir com o agente pelo playground do console doGoogle Cloud . Para mais informações, consulte Suporte do Playground.

Se você quiser usar esses recursos, implemente os seguintes endpoints:

• /api/reasoning_engine: usado para disponibilizar consultas enviadas à API REST reasoningEngines/query ou aos métodos de consulta síncronos e assíncronos do SDK do Python.
• /api/stream_reasoning_engine: usado para exibir consultas enviadas à API REST reasoningEngines/streamQuery ou aos métodos de consulta de streaming do SDK para Python.

Métodos de classe e modos de execução

Ao implantar um contêiner personalizado, é necessário declarar os métodos de classe compatíveis na lista classMethods da especificação da implantação. Esses métodos de classe correspondem às operações que você definiu ao desenvolver o agente (consulte Registrar métodos personalizados e Consultar o agente usando operações compatíveis). Cada método declarado tem um name e um api_mode que determinam como ele é roteado:

Modo de API	Tipo de execução	Endpoint de roteamento
"" (string vazia) ou "async"	Unário (solicitação-resposta)	/api/reasoning_engine
"stream" ou "async_stream"	Streaming	/api/stream_reasoning_engine

Quando um cliente invoca um método, o serviço do Agent Runtime envia uma solicitação POST para o endpoint de roteamento correspondente no contêiner. O corpo JSON da solicitação contém o campo class_method (correspondente ao nome do método) e o campo input.

Métodos necessários para integração

Para usar o SDK para Python ou o playground do console do Google Cloud , seu contêiner precisa implementar os métodos específicos que essas integrações esperam:

• Consulta padrão do SDK: requer query (modo "" ou "async") e stream_query (modo "stream" ou "async_stream").
• Playground: requer stream_query (modo "stream" ou "async_stream").

Integração do ADK

Se você estiver implantando um agente criado com o Kit de Desenvolvimento de Agente (ADK), poderá criar seu próprio contêiner de proxy na linguagem de programação e no framework de servidor de sua escolha. Para oferecer suporte ao conjunto completo de recursos do ADK, seu contêiner precisa implementar os métodos definidos pelo contrato do ADK. Para mais informações, consulte Usar um agente do ADK e Registrar e gerenciar um agente do ADK.

Consulte o modelo AdkApp como uma implementação de referência. Para mais informações, consulte a documentação de referência do AdkApp e o código-fonte do AdkApp.

Se você quiser que o Google processe as atualizações automaticamente quando você atualizar sua versão do ADK, use as ferramentas fornecidas pelo ADK para implantar agentes em vez de escrever seu próprio servidor de API. Para mais informações, consulte a documentação de implantação do ADK.

Especificações da API

Tanto /api/reasoning_engine quanto /api/stream_reasoning_engine recebem solicitações HTTP POST com um corpo JSON que contém os seguintes campos:

• class_method (string): o nome do método do agente subjacente a ser invocado (por exemplo, query ou stream_query).
• input (objeto JSON): os argumentos a serem transmitidos para o método de classe especificado.

/api/reasoning_engine (unário)

• Método de solicitação: POST
• Corpo da solicitação: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
• Resposta: um objeto JSON que contém a saída da execução do agente. json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (streaming)

• Método de solicitação: POST
• Corpo da solicitação: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
• Resposta: um stream de JSON delimitado por linha (ndjson), em que cada linha é um bloco codificado em JSON da resposta. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

Exemplo de servidor de API (Python)

Confira a seguir um exemplo de servidor FastAPI em Python que implementa o contrato de ambiente de execução da plataforma de agentes. Esse servidor envolve um agente de exemplo (SimpleAgent) e processa o roteamento e a codificação. É possível substituir SimpleAgent pela sua própria implementação de agente.

Para executar este exemplo, verifique se você tem fastapi, uvicorn e pydantic instalados.

import inspect
import json
import logging
import os
import uvicorn
from fastapi import FastAPI, encoders, responses
from pydantic import BaseModel

app = FastAPI()

# Define the request body structure
class QueryRequest(BaseModel):
    input: dict | None = None
    class_method: str

# Example Agent implementation
class SimpleAgent:
    def query(self, message: str) -> str:
        return f"Echo: {message}"

    async def stream_query(self, message: str):
        words = message.split()
        for word in words:
            yield {"output": word + " "}

agent = SimpleAgent()

def _encode_chunk_to_json(chunk):
    """Encodes a chunk to a JSON string with a newline."""
    try:
        json_chunk = encoders.jsonable_encoder(chunk)
        return json.dumps(json_chunk) + "\n"
    except Exception:
        logging.exception("Failed to encode chunk")
        return None

async def json_generator(output):
    async for chunk in output:
        encoded_chunk = _encode_chunk_to_json(chunk)
        if encoded_chunk is None:
            break
        yield encoded_chunk

async def _invoke_callable_or_raise(invocation_callable, invocation_payload):
    if inspect.iscoroutinefunction(invocation_callable):
        return await invocation_callable(**invocation_payload)
    else:
        return invocation_callable(**invocation_payload)

@app.post("/api/reasoning_engine")
async def query_endpoint(request: QueryRequest) -> responses.JSONResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.JSONResponse(
            status_code=400,
            content={"error": f"Method {request.class_method} not found on agent"}
        )

    output = await _invoke_callable_or_raise(method, request.input or {})

    try:
        json_serialized_content = encoders.jsonable_encoder({"output": output})
    except ValueError as encoding_error:
        logging.exception("Failed to JSON-encode response: %s", encoding_error)
        raise encoding_error
    return responses.JSONResponse(content=json_serialized_content)

@app.post("/api/stream_reasoning_engine")
async def stream_query_endpoint(request: QueryRequest) -> responses.StreamingResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.StreamingResponse(
            content=iter([json.dumps({"error": f"Method {request.class_method} not found"})]),
            status_code=400,
            media_type="application/json"
        )

    output = await _invoke_callable_or_raise(method, request.input or {})
    return responses.StreamingResponse(
        content=json_generator(output),
        media_type="application/json",
    )

if __name__ == "__main__":
    # The container must listen on 0.0.0.0 and port 8080
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))