HTTP Client

O HttpClient é um cliente HTTP síncrono (httpx) com retry automático, mode IP fixo (Azure APIM), mTLS por certificado de cliente 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),
)

# Com certificado de cliente (mTLS)
client = HttpClient(ctx, cert_content=secrets["CERTIFICADO_PEM"])

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
`cert_content`	`str \| None`	`None`	Certificado de cliente em PEM (chave privada + certificado) para mTLS. Ver mTLS

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 (pode vir "" se a resposta for binária)
resposta.content         # bytes — corpo bruto, sem decodificação
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

Para respostas binárias (ex.: PDF, imagem, anexo), use resposta.content em vez de resposta.text — text não é confiável nesses casos (pode vir vazio ou corrompido) porque não passa por decodificação:

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

with open("boleto.pdf", "wb") as f:
    f.write(resposta.content)

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.

mTLS — certificado de cliente

Para APIs que exigem certificado de cliente (mTLS), informe o PEM em cert_content ao criar o cliente:

from actions_sdk.features.http import HttpClient

# Secret do tipo CERT — chave privada + certificado(s) num único PEM
client = HttpClient(ctx, cert_content=secrets["CERTIFICADO_PEM"])

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

O certificado vale para todas as requisições desse cliente, e a chamada continua com retry, sanitização de logs e tracing como qualquer outra. Pode ser combinado com use_fixed_ip=True.

Cada cliente carrega um certificado. Para dois endpoints com certificados diferentes, crie um cliente para cada:

client_a = HttpClient(ctx, cert_content=secrets["CERT_PARCEIRO_A"])
client_b = HttpClient(ctx, cert_content=secrets["CERT_PARCEIRO_B"])

Um PEM inválido falha na hora de criar o cliente, e não durante a requisição:

Caso	Resultado
PEM sem chave privada	`ValueError`
PEM inválido ou incompleto	`ssl.SSLError`
`cert_content=""`	`ValueError`
`cert_content=None` (padrão)	Cliente comum, sem certificado

note

O certificado não vai em headers, query params nem corpo — não aparece em logs nem no trace.

CertificateClient

Depreciado

Use HttpClient(ctx, cert_content=...) — ver mTLS.

CertificateClient continua funcional e não será removido sem aviso, mas usa requests diretamente: não tem retry, não sanitiza logs e não emite o span http_client — a chamada não aparece no tracing da action.

Ao migrar, note que não é uma troca direta: o retorno passa a ser HttpResult e os erros passam a ser SDKHttpError em vez de requests.HTTPError.

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(),
        )

Inicialização​

Métodos HTTP​

get(url)​

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

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

request(method, url) — genérico​

HttpResult​

RetryPolicy​

SDKHttpError​

Modo IP Fixo (Azure APIM)​

mTLS — certificado de cliente​

CertificateClient​

Exemplo completo​