HTTP Client

O HttpClient é um cliente HTTP síncrono (httpx) com retry automático, mode IP fixo (Azure APIM) e tracing integrado. Use para integrar com APIs externas e sistemas internos.

Inicialização

from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.features.http import HttpClient

ctx: ActionContext

client = HttpClient(ctx)

# Com timeout e política de retry customizados
from actions_sdk.features.http.client import RetryPolicy

client = HttpClient(
    ctx,
    default_timeout=15.0,
    default_retry_policy=RetryPolicy(max_attempts=2),
)

Parâmetros do construtor:

Parâmetro	Tipo	Padrão	Descrição
ctx	BaseContext	—	Contexto de execução (ActionContext ou TaskContext)
default_timeout	float	30.0	Timeout padrão em segundos
default_retry_policy	RetryPolicy | None	RetryPolicy()	Política de retry padrão

---

Métodos HTTP

Todos os métodos retornam HttpResult ou lançam SDKHttpError.

get(url)

resposta = client.get("https://api.exemplo.com/dados")
resposta = client.get("https://api.exemplo.com/dados", timeout=5.0)
resposta = client.get("https://api.exemplo.com/dados", params={"id": "123"})

post(url, json=..., data=...)

resposta = client.post(
    "https://api.exemplo.com/criar",
    json={"nome": "Alice", "email": "alice@exemplo.com"},
)

put(url), patch(url), delete(url)

resposta = client.put("https://api.exemplo.com/itens/1", json={"status": "ativo"})
resposta = client.patch("https://api.exemplo.com/itens/1", json={"status": "inativo"})
resposta = client.delete("https://api.exemplo.com/itens/1")

request(method, url) — genérico

resposta = client.request(
    method="POST",
    url="https://api.exemplo.com/endpoint",
    json={"chave": "valor"},
    headers={"Authorization": "Bearer meu-token"},
    timeout=10.0,
)

Parâmetros comuns a todos os métodos:

Parâmetro	Tipo	Padrão	Descrição
url	str	—	URL da requisição
params	dict | None	None	Query parameters
headers	dict | None	None	Headers HTTP customizados
json	dict | list | None	None	Corpo JSON
data	dict | str | bytes | None	None	Corpo form/raw
timeout	float | None	default_timeout	Timeout em segundos
follow_redirects	bool	True	Seguir redirecionamentos
retry_policy	RetryPolicy | None	default_retry_policy	Política de retry por requisição
expected_status	set[int] | None	None	Status codes considerados bem-sucedidos (não levantam SDKHttpError)
sensitive_json_keys	list[str] | None	None	Chaves a mascarar nos logs de tracing

---

HttpResult

Objeto retornado por todas as requisições bem-sucedidas:

resposta = client.get("https://api.exemplo.com/dados")

resposta.status_code     # int — ex: 200
resposta.headers         # dict[str, str]
resposta.text            # str — corpo como texto
resposta.duration_ms     # float — duração total em ms
resposta.request_url     # str — URL final após redirecionamentos
resposta.request_method  # str — método HTTP usado

# Parse JSON
dados = resposta.json()  # dict | list — lança ValueError se não for JSON válido

# Helpers de status
resposta.is_success()       # True se 2xx
resposta.is_client_error()  # True se 4xx
resposta.is_server_error()  # True se 5xx

---

RetryPolicy

Configura o comportamento de retry automático:

from actions_sdk.features.http.client import RetryPolicy

policy = RetryPolicy(
    max_attempts=3,           # Total de tentativas (incluindo a primeira)
    base_delay_ms=0,          # Delay base em ms (backoff exponencial)
    max_delay_ms=1000,        # Delay máximo em ms
    jitter_factor=0.1,        # Fator de jitter (0.0 a 1.0)
    retry_on_status_codes={429, 500, 502, 503, 504},  # Códigos com retry
)

Campo	Padrão	Descrição
max_attempts	3	Número máximo de tentativas
base_delay_ms	0	Delay base para backoff exponencial
max_delay_ms	1000	Delay máximo em milissegundos
jitter_factor	0.1	Variação aleatória para evitar thundering herd
retry_on_status_codes	{429, 500, 502, 503, 504}	Status HTTP que ativam retry

---

SDKHttpError

Exceção lançada quando a requisição falha após todos os retries:

from actions_sdk.features.http.client import SDKHttpError

try:
    resposta = client.get("https://api.exemplo.com/dados")
except SDKHttpError as erro:
    erro.status_code           # int | None — código HTTP
    erro.error_type            # str — "timeout", "connection", "http_4xx", "http_5xx", "unknown"
    erro.message               # str — mensagem de erro
    erro.method                # str — método HTTP
    erro.url                   # str — URL da requisição
    erro.response_body_snippet # str | None — trecho do corpo da resposta
    erro.duration_ms           # float | None — duração em ms

Tipos de erro (error_type):

Valor	Quando ocorre
"timeout"	Timeout da requisição atingido
"connection"	Falha de conexão / rede
"http_4xx"	Status 4xx após todos os retries
"http_5xx"	Status 5xx após todos os retries
"unknown"	Outros erros

---

Modo IP Fixo (Azure APIM)

Para APIs que exigem IP fixo via Azure APIM:

resposta = client.post(
    "https://apim.exemplo.com/endpoint",
    json={"chave": "valor"},
    use_fixed_ip=True,
    x_redirect_key="minha-chave-de-redirect",
)

Requer ctx.fixed_ip_subscription_key configurado no motor.

---

CertificateClient

Para requisições HTTPS que exigem autenticação por certificado de cliente (mTLS):

from actions_sdk.features.http.certificate_client import CertificateClient

cert_content = "-----BEGIN PRIVATE KEY-----\n..."  # Conteúdo PEM como string

client_cert = CertificateClient(cert_content=cert_content)
resposta = client_cert.get("https://api-segura.exemplo.com/dados")
dados = resposta.json()

Métodos disponíveis: get(), post(), put(), delete(), request()

Exceções:

• requests.RequestException — falha na requisição
• ValueError — parsing do certificado falhou

note

CertificateClient usa requests internamente (não httpx). Não suporta RetryPolicy nem SDKHttpError.

---

Exemplo completo

Baseado nas actions http_get, http_request e retry_policy_homol do agente de homologação:

import traceback

from actions_sdk.core.types.action_context import ActionContext
from actions_sdk.core.types.responses import ResponseStatus, ResponseToAgent
from actions_sdk.features.conversational.state import StateManager
from actions_sdk.features.http import HttpClient
from actions_sdk.features.http.client import SDKHttpError
from actions_sdk.features.tracing import ActionSpan

ctx: ActionContext

TIMEOUT = 10.0
CHAVE_PROGRESSO = "action.http.progresso"


def consultar_api(url: str, method: str = "GET"):
    span = ActionSpan(ctx)
    state = StateManager(ctx)

    span.event("action.http.inicio", {"url": url, "method": method})

    try:
        client = HttpClient(ctx)
        resposta = client.request(method=method, url=url, timeout=TIMEOUT)

        status_code = resposta.status_code
        preview = str(resposta.text)[:300]
        state.update(CHAVE_PROGRESSO, "SUCESSO ✓")

        span.event("action.http.sucesso", {"status": status_code})

        return ResponseToAgent(
            status=ResponseStatus.SUCCESS,
            instruction=(
                f"HttpClient.{method.lower()}() — SUCESSO ✓\n"
                f"URL: {url}\n"
                f"Status: {status_code}\n"
                f"Body: {preview}"
            ),
            state_updates=state.get_full_state(),
        )

    except SDKHttpError as erro:
        state.update(CHAVE_PROGRESSO, "HTTP_ERROR ✗")
        span.event("action.http.sdk_error", {"status": erro.status_code, "tipo": erro.error_type})
        return ResponseToAgent(
            status=ResponseStatus.ERROR,
            instruction=(
                f"Falha HTTP ({erro.error_type})\n"
                f"Status: {erro.status_code}\n"
                f"Mensagem: {erro.message}"
            ),
            state_updates=state.get_full_state(),
        )

    except Exception as erro:
        state.update(CHAVE_PROGRESSO, "ERRO ✗")
        span.event("action.http.erro", {"erro": str(erro)})
        return ResponseToAgent(
            status=ResponseStatus.ERROR,
            instruction=f"Erro inesperado: {traceback.format_exc()}",
            state_updates=state.get_full_state(),
        )