O Agent Runtime foi projetado para ser independente da estrutura do aplicativo. Se você optar por implantar o agente usando um contêiner personalizado ou um Dockerfile, o contêiner precisará obedecer ao contrato de execução para veicular consultas.
Para mais informações sobre 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)
O contêiner pode expor qualquer endpoint HTTP personalizado. É possível invocar esses endpoints personalizados enviando solicitações à API subjacente do agente implantado. Para mais informações, consulte Usar agentes implantados pela API subjacente.
Embora esses endpoints sejam opcionais no nível da API, a implementação deles ativa os seguintes recursos de integração:
- Suporte ao SDK do Python: a implementação de
/api/reasoning_enginee/api/stream_reasoning_enginepermite usar o agente implantado pelo SDK da Agent Platform para Python. - Suporte ao playground: a implementação de
/api/stream_reasoning_engineé necessária se você quiser interagir com o agente pelo Google Cloud playground do console. Para mais informações, consulte Suporte ao playground.
Se você quiser usar esses recursos, implemente os seguintes endpoints:
/api/reasoning_engine: usado para veicular consultas enviadas à API RESTreasoningEngines/queryou aos métodos de consulta síncronos e assíncronos do SDK do Python./api/stream_reasoning_engine: usado para veicular consultas enviadas à API RESTreasoningEngines/streamQueryou aos métodos de consulta de streaming do SDK do Python.
Métodos de classe e modos de execução
Ao implantar um contêiner personalizado, declare os métodos de classe compatíveis na lista classMethods da especificação de implantação. Esses métodos de classe correspondem às operações definidas 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 determina 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 (que corresponde ao nome do método) e o campo input.
Métodos necessários para integração
Para usar o SDK do Python ou o Google Cloud playground do console, o contêiner precisa implementar os métodos específicos esperados por essas integrações:
- Consulta padrão do SDK: requer
query(modo""ou"async") estream_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 na estrutura de servidor de sua escolha. Para oferecer suporte ao conjunto completo de recursos do ADK, o 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 atualizações automaticamente ao 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
Os endpoints /api/reasoning_engine e /api/stream_reasoning_engine recebem solicitações POST HTTP com um corpo JSON contendo os seguintes campos:
class_method(string): o nome do método do agente subjacente a ser invocado (por exemplo,queryoustream_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 contendo 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 fluxo 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 um servidor FastAPI em Python que implementa o contrato de execução da Agent Platform. Esse servidor envolve um agente de exemplo (SimpleAgent) e processa o roteamento e a codificação. Você pode substituir SimpleAgent pela sua própria implementação de agente.
Para executar esse 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)))