Quando não há API: o banco

EJUD e DCP são anteriores à ideia de webservice. Não há endpoint, não há WSDL, não há chave para pedir por e-mail. O dado existe, é consultável, e a única porta é o banco de dados.

Isso muda três coisas de uma vez: quem autoriza, o que a autorização alcança, e onde mora o segredo. Este capítulo trata das três, e o código vem de um projeto real de conferência de dados que roda contra o EJUD — não de um exemplo inventado para o curso.

Primeiro o chamado, depois o código

O erro de ordem mais caro desta unidade é escrever a Tool e depois descobrir o que a credencial alcança. Faça ao contrário.

O que pedir à TI, e por quê:

1. Um usuário nominal, só seu. Não a credencial de aplicação que a equipe usa, não uma compartilhada com a vara. Se um dia for preciso saber o que essa ferramenta consultou, o log tem que apontar para uma pessoa.

2. Somente leitura — e leitura só do que você precisa. "Somente leitura" é a parte fácil e a parte que não protege quase nada: ela impede que você escreva, não que você leia o Tribunal inteiro. A parte que protege é a segunda: peça acesso a views, não a tabelas.

3. Uma view que já venha filtrada. Este é o pedido que a maioria das pessoas não faz, e é o único que resolve o problema do capítulo anterior. Em vez de SELECT na tabela de processos, peça uma view que já traga apenas o seu órgão julgador e apenas o que não é sigiloso. O filtro passa a existir do lado do banco, aplicado por quem administra o banco, e nenhum erro seu em Python consegue desfazê-lo.

4. Acesso pela VPN, com a mesma exigência de sempre. Se a rede exige token, a sua ferramenta também exige. Uma ferramenta que funciona fora da VPN é uma ferramenta que abriu um caminho que não existia.

O driver: oracledb em modo thin

Uma boa notícia para um curso que roda em Windows nativo:

pip install oracledb

E acabou. O oracledb moderno tem um modo thin, escrito em Python puro, que fala o protocolo do Oracle direto pela rede. Nada de Instant Client, nada de variável PATH apontando para DLL, nada de instalador que pede administrador — que é exatamente o que a máquina do Tribunal não vai te dar.

Isso não é detalhe de conforto. O antecessor (cx_Oracle) exigia bibliotecas nativas instaladas, e "instale o Instant Client" é onde a maior parte das tentativas morria numa máquina sem administrador.

O segredo não entra no arquivo

Aqui está o padrão, e ele é curto:

# config.yaml — este arquivo pode ir para o repositório {: #configyaml-este-arquivo-pode-ir-para-o-repositório }
oracle:
  host: ""
  port: 1521
  service_name: ""
  user: ""
  password_env: ORACLE_PASSWORD    # o NOME da variável, não a senha

import os
import oracledb

def conectar(cfg: dict) -> oracledb.Connection:
    senha = os.environ.get(cfg["password_env"], "")
    if not senha:
        raise SystemExit(
            f"variável de ambiente {cfg['password_env']} não está definida. "
            "Defina-a antes de rodar; ela não fica em arquivo."
        )
    return oracledb.connect(
        user=cfg["user"],
        password=senha,
        host=cfg["host"],
        port=int(cfg["port"]),
        service_name=cfg["service_name"],
    )

O arquivo de configuração guarda o nome da variável de ambiente, nunca o valor. O config.yaml pode ser versionado, enviado por e-mail, colado num chamado — não tem nada dentro. A senha vive só na sessão em que você a definiu:

$env:ORACLE_PASSWORD = "..."

E repare no raise SystemExit com o nome da variável no texto. Falha cedo e falha explicando. A alternativa — senha vazia indo para o connect — produz um ORA-01017: invalid username/password que manda você conferir o usuário, que está certo.

Descobrir o formato sem ler o dado

Antes de escrever qualquer consulta, você precisa saber quais colunas existem. A forma cara é SELECT * com um limite. A forma barata:

def colunas(conn, tabela: str) -> list[str]:
    cur = conn.cursor()
    cur.execute(f"SELECT * FROM {tabela} WHERE 1=0")
    return [d[0].upper() for d in cur.description]

WHERE 1=0 é falso para toda linha. O banco resolve o plano, monta a descrição do resultado e devolve zero linhas — mas cursor.description já traz os nomes e tipos de todas as colunas. Você aprendeu o formato da tabela sem ter lido um único dado de processo.

Numa Tool que vai conversar com um modelo de linguagem, isso é mais do que economia: é a única introspecção que você pode rodar sem se perguntar se o resultado pode vazar.

As cinco formas, agora em SQL

As cinco formas da Unidade 3 não eram sobre Elasticsearch. Eram sobre perguntas, e elas atravessam para o SQL sem tradução:

E a lição da forma 1 vale igual: COUNT(*) devolve um número e responde "quantos". Uma Tool que só sabe listar vai responder "quais" mesmo quando lhe perguntam "quantos" — e vai responder vinte.

A diferença entre um painel e uma Tool

Este é o ponto do capítulo. Ele é sutil e ele é a razão de a unidade existir.

Considere uma consulta que filtra por período, com o valor vindo de um campo de texto:

# Painel: eu digito "2026-05" numa caixa e clico. {: #painel-eu-digito-2026-05-numa-caixa-e-clico }
where = f" WHERE TO_CHAR({col_data}, 'YYYY-MM') = '{periodo}'"
cur.execute(f"SELECT COUNT(*) FROM {tabela}{where}")

Num painel que você dirige, isso é aceitável, e milhares de scripts internos fazem assim. O usuário é você, o alcance é o seu, e se você digitar bobagem o erro é seu e aparece na sua tela.

Agora a mesma linha dentro de uma Tool. O que muda não é o SQL: muda quem escolhe o valor do período.

Numa Tool, o argumento é escolhido por um modelo de linguagem, a partir de um texto. E esse texto pode ser um documento, uma petição, um e-mail — conteúdo que você não escreveu e não revisou. O modelo é obediente e não distingue instrução de dado. Basta que o valor escolhido seja

2026-05' OR '1'='1

para que a sua consulta filtrada deixe de ser filtrada — e a mesma técnica que derruba o filtro de período derruba o filtro de órgão julgador e o de sigilo.

Não é hipótese exótica. É a consequência direta de ter transformado uma entrada que era sua numa entrada que é de terceiros.

A forma certa é a de sempre, e é mais curta:

def contar_no_periodo(conn, ano_mes: str) -> int:
    cur = conn.cursor()
    cur.execute(
        """
        SELECT COUNT(*)
          FROM vw_processos_meu_orgao
         WHERE TO_CHAR(dth_referencia, 'YYYY-MM') = :ano_mes
        """,
        ano_mes=ano_mes,
    )
    return int(cur.fetchone()[0])

:ano_mes é um bind. O valor viaja por fora do texto da consulta, e o banco nunca o interpreta como SQL. Aquele 2026-05' OR '1'='1 vira, aqui, simplesmente um período que não existe: a resposta é zero.

Repare no que ficou fixo no código e no que ficou variável:

• Fixos, escritos por você: o nome da view, os nomes das colunas, a estrutura da consulta.
• Variável, vindo do modelo: um valor, e só valores.

Nome de tabela e nome de coluna não podem ser bind em nenhum banco. Isso não é limitação: é o desenho certo. Se a sua Tool precisa que o modelo escolha a tabela, o problema não é o bind — é que você está deixando o modelo escolher o alcance da consulta. Se for mesmo necessário, a escolha é entre nomes de uma lista que você escreveu:

TABELAS = {"processos": "vw_processos_meu_orgao",
           "movimentos": "vw_movimentos_meu_orgao"}

def consultar(conjunto: str, ...):
    if conjunto not in TABELAS:
        raise ValueError(f"conjunto desconhecido: {conjunto}")
    tabela = TABELAS[conjunto]   # nome vindo do seu dicionário, nunca do modelo

O sigilo se filtra no banco, não no Python

Na Unidade 2, o sigilo não era problema seu: o CNJ já entrega o índice sem o que é sigiloso, e o TETO_SIGILO = 0 das Tools era cinto de segurança sobre um dado que já vinha limpo.

Aqui é o contrário. O banco interno tem tudo, e vai te entregar tudo que a sua credencial alcança — porque você está autorizado a ver. A pergunta deixou de ser "eu posso ver?" e passou a ser "isto pode ir para o modelo?".

E a resposta é não. O modelo é um terceiro. Mesmo rodando local no Ollama, na sua máquina, ele é um destino a mais para onde o dado foi copiado — e no dia em que alguém trocar o api_base para um serviço remoto, o destino muda sem que a Tool mude uma linha.

Duas regras, nesta ordem:

1. O filtro mora na view. Se a TI te deu vw_processos_meu_orgao já sem sigilo, nenhum erro de Python desfaz isso. É a única proteção que sobrevive a você.

2. Se não houver view, o filtro mora no WHERE. Nunca no pandas depois.

# Certo: o dado sigiloso nunca chega ao processo que fala com o modelo. {: #certo-o-dado-sigiloso-nunca-chega-ao-processo-que-fala-com-o-modelo }
cur.execute(
    "SELECT numero_processo, dth_ajuizamento "
    "  FROM processos "
    " WHERE nivel_sigilo = 0 "
    "   AND cod_orgao = :orgao",
    orgao=orgao,
)

# Errado, e passa em todo teste que você fizer. {: #errado-e-passa-em-todo-teste-que-você-fizer }
df = pd.read_sql("SELECT * FROM processos", conn)
df = df[df.nivel_sigilo == 0]

O segundo funciona. O resultado é idêntico. E ele já trouxe todos os processos sigilosos para dentro da memória do mesmo processo que monta o prompt — onde um print de depuração, um traceback ou um df.head() num log basta para publicá-los.

Juntando: uma Tool inteira

import os
import oracledb

# Conexão única, reaproveitada. Abrir conexão por chamada é lento e enche o {: #conexão-única-reaproveitada-abrir-conexão-por-chamada-é-lento-e-enche-o }
# banco de sessões — e o agente chama a Tool várias vezes por pergunta. {: #banco-de-sessões-e-o-agente-chama-a-tool-várias-vezes-por-pergunta }
_conexao: oracledb.Connection | None = None


def _conn() -> oracledb.Connection:
    global _conexao
    if _conexao is None:
        senha = os.environ.get("ORACLE_PASSWORD", "")
        if not senha:
            raise SystemExit("ORACLE_PASSWORD não definida.")
        _conexao = oracledb.connect(
            user=os.environ["ORACLE_USER"],
            password=senha,
            host=os.environ["ORACLE_HOST"],
            port=1521,
            service_name=os.environ["ORACLE_SERVICE"],
        )
    return _conexao


def contar_processos_por_assunto(assunto: str, ano: int) -> dict:
    """Quantos processos do meu órgão, por assunto, num ano.

    Args:
        assunto: nome do assunto, como aparece na tabela do CNJ.
        ano: ano de ajuizamento, com quatro dígitos.
    """
    cur = _conn().cursor()
    cur.execute(
        """
        SELECT COUNT(*)
          FROM vw_processos_meu_orgao
         WHERE assunto_nome = :assunto
           AND EXTRACT(YEAR FROM dth_ajuizamento) = :ano
        """,
        assunto=assunto,
        ano=ano,
    )
    return {"assunto": assunto, "ano": ano, "total": int(cur.fetchone()[0])}

Vinte e poucas linhas, e cada decisão do capítulo está em uma delas: senha vinda do ambiente, view em vez de tabela, binds em vez de f-string, COUNT em vez de lista, e um retorno de três campos em vez de um DataFrame.

A docstring não é documentação: é o que o modelo lê para decidir se chama esta Tool e com que argumentos. Se ela disser "consulta processos", o modelo vai chamá-la para perguntas de listar também — e ela não lista.

Antes de apontar para produção

Uma lista curta, para rodar na ordem:

1. SELECT * FROM sua_view WHERE 1=0 conecta e devolve as colunas que você espera?
2. SELECT COUNT(*) FROM sua_view bate com o que a tela do sistema mostra? Se der mais, sua credencial alcança mais que você — volte ao chamado.
3. Existe alguma linha com nivel_sigilo > 0 na view? Se existir, o filtro não está onde você pensou.
4. Alguma Tool sua monta SQL com f-string a partir de argumento? Troque por bind antes de conectar o agente.
5. Alguma Tool devolve coluna que você não listou à mão?

Os itens 2 e 3 são os que realmente importam, e são os únicos que não dá para verificar lendo o código.

Próximo: Só os meus processos, onde a credencial deixa de ser só uma senha e passa a ser a resposta para "quais são os meus".