Pular para conteúdo

shared

Modulo compartilhado: utilitarios, clientes HTTP, validadores e schemas base.

APIError

APIError(message, status_code=None, endpoint=None, details=None)

Bases: MCPFiscalError

Erro retornado por uma API externa (SEFAZ, Receita Federal, etc.).

MCPFiscalError

MCPFiscalError(message, code=None, details=None)

Bases: Exception

Exceção base para todos os erros do MCP Fiscal Brasil.

NotFoundError

NotFoundError(resource, identifier)

Bases: MCPFiscalError

Recurso não encontrado na API ou base de dados.

RateLimitError

RateLimitError(endpoint, retry_after=None)

Bases: MCPFiscalError

Limite de requisições excedido para um endpoint.

TimeoutError

TimeoutError(endpoint, timeout_seconds)

Bases: MCPFiscalError

Timeout ao comunicar com serviço externo.

ValidationError

ValidationError(field, value, reason)

Bases: MCPFiscalError

Dado inválido fornecido pelo usuário (CPF, CNPJ, chave NFe, etc.).

XMLParseError

XMLParseError(message, raw_content=None)

Bases: MCPFiscalError

Erro ao processar XML de NFe, NFSe ou SPED.

BaseResponse

Bases: BaseModel

Resposta base para todas as ferramentas MCP.

ErrorResponse

Bases: BaseResponse

Resposta de erro padronizada.

format_cnpj

format_cnpj(cnpj, remover_mascara=False)

Formata um CNPJ com ou sem mascara.

Se remover_mascara=True, retorna apenas os 14 digitos numericos. Caso contrario, retorna no formato XX.XXX.XXX/XXXX-XX.

format_cpf

format_cpf(cpf, remover_mascara=False)

Formata um CPF com ou sem mascara.

Se remover_mascara=True, retorna apenas os 11 digitos numericos. Caso contrario, retorna no formato XXX.XXX.XXX-XX.

validate_chave_nfe

validate_chave_nfe(chave)

Valida uma chave de acesso de NFe/NFCe (44 digitos).

A chave e composta por: cUF(2) + AAMM(4) + CNPJ(14) + mod(2) + serie(3) + nNF(9) + tpEmis(1) + cNF(8) + cDV(1) = 44 digitos.

Verifica apenas o digito verificador (modulo 11).

validate_cnpj

validate_cnpj(cnpj)

Valida um CNPJ brasileiro.

Aceita formatos com ou sem mascara (XX.XXX.XXX/XXXX-XX ou XXXXXXXXXXXXXX).

validate_cpf

validate_cpf(cpf)

Valida um CPF brasileiro.

Aceita formatos com ou sem mascara (XXX.XXX.XXX-XX ou XXXXXXXXXXX). Retorna False para CPFs com todos os digitos iguais (ex: 111.111.111-11).

constants

Constantes fiscais brasileiras: UFs, CFOP, CST, NCM e códigos SEFAZ.

exceptions

Hierarquia de exceções do MCP Fiscal Brasil.

MCPFiscalError

MCPFiscalError(message, code=None, details=None)

Bases: Exception

Exceção base para todos os erros do MCP Fiscal Brasil.

APIError

APIError(message, status_code=None, endpoint=None, details=None)

Bases: MCPFiscalError

Erro retornado por uma API externa (SEFAZ, Receita Federal, etc.).

RateLimitError

RateLimitError(endpoint, retry_after=None)

Bases: MCPFiscalError

Limite de requisições excedido para um endpoint.

ValidationError

ValidationError(field, value, reason)

Bases: MCPFiscalError

Dado inválido fornecido pelo usuário (CPF, CNPJ, chave NFe, etc.).

NotFoundError

NotFoundError(resource, identifier)

Bases: MCPFiscalError

Recurso não encontrado na API ou base de dados.

TimeoutError

TimeoutError(endpoint, timeout_seconds)

Bases: MCPFiscalError

Timeout ao comunicar com serviço externo.

XMLParseError

XMLParseError(message, raw_content=None)

Bases: MCPFiscalError

Erro ao processar XML de NFe, NFSe ou SPED.

AuthError

AuthError(service, reason)

Bases: MCPFiscalError

Erro de autenticação com serviço externo (certificado digital, token, etc.).

http_client

Cliente HTTP assincrono com retry, backoff e integracao com rate limiter.

FiscalHTTPClient

FiscalHTTPClient(base_url, timeout=DEFAULT_TIMEOUT, max_retries=3, backoff_factor=1.5, rate_limiter=None, headers=None)

Cliente HTTP assincrono com: - Retry automático com backoff exponencial - Integracao com SlidingWindowRateLimiter - Mapeamento de erros HTTP para excecoes fiscais - Logging estruturado

close async

close()

Fecha a conexao HTTP.

get async

get(path, params=None, headers=None, rate_limit_key=None)

Executa GET com retry e rate limiting.

post async

post(path, json=None, data=None, headers=None, rate_limit_key=None)

Executa POST com retry e rate limiting.

rate_limiter

Rate limiter de janela deslizante por endpoint.

SlidingWindowRateLimiter

SlidingWindowRateLimiter(max_requests, window_seconds)

Implementa um rate limiter de janela deslizante por chave (ex: endpoint).

Permite configurar N requisicoes por janela de tempo (em segundos). Thread-safe via asyncio.Lock.

acquire async

acquire(key)

Tenta adquirir um slot para a chave informada.

Levanta RateLimitError se o limite for excedido.

remaining

remaining(key)

Retorna quantas requisicoes ainda podem ser feitas na janela atual.

reset

reset(key)

Limpa o historico de uma chave (util para testes).

schemas

Schemas Pydantic base para respostas de API.

BaseResponse

Bases: BaseModel

Resposta base para todas as ferramentas MCP.

ErrorResponse

Bases: BaseResponse

Resposta de erro padronizada.

PaginatedResponse

Bases: BaseResponse, Generic[T]

Resposta paginada generica.

Endereco

Bases: BaseModel

Endereco padrao brasileiro.

validators

Validadores de documentos fiscais brasileiros (CPF, CNPJ, chave NFe) e caminhos de arquivo.

validar_caminho_arquivo

validar_caminho_arquivo(path, *, label='Arquivo')

Normaliza e valida um caminho de arquivo contra path traversal e injeção.

Resolve o caminho para seu valor real (sem .., links simbólicos, etc.) e rejeita entradas com padrões suspeitos de traversal antes de abrir o arquivo. Não restringe a um diretório fixo - o operador pode apontar arquivos em qualquer local do sistema - o objetivo é bloquear injeção de caminho controlada por dados externos não confiáveis (ex: input de LLM).

Parameters:

Name Type Description Default
path str | Path

Caminho informado pelo chamador (string ou Path).

required
label str

Rótulo descritivo para mensagens de erro (ex: "Arquivo SPED").

'Arquivo'

Returns:

Type Description
Path

Path resolvido e validado.

Raises:

Type Description
ValueError

Quando o caminho contém componentes suspeitos de traversal ou se o arquivo não existe ou não é legível.

validate_cpf

validate_cpf(cpf)

Valida um CPF brasileiro.

Aceita formatos com ou sem mascara (XXX.XXX.XXX-XX ou XXXXXXXXXXX). Retorna False para CPFs com todos os digitos iguais (ex: 111.111.111-11).

validate_cnpj

validate_cnpj(cnpj)

Valida um CNPJ brasileiro.

Aceita formatos com ou sem mascara (XX.XXX.XXX/XXXX-XX ou XXXXXXXXXXXXXX).

validate_cnpj_alfanumerico

validate_cnpj_alfanumerico(cnpj)

Valida um CNPJ alfanumérico (IN RFB 2.229/2024, vigência jul/2026).

As 12 primeiras posições podem conter letras maiúsculas (A-Z) ou dígitos. As 2 últimas são dígitos verificadores numéricos calculados pelo módulo 11, usando o valor ASCII de cada caractere subtraído de 48.

Esta função NÃO aceita máscaras (pontos, barra, traço) nem letras minúsculas. Aceita CNPJs numéricos puros (14 dígitos sem máscara) como caso especial, pois o algoritmo é compatível com ambos os formatos. Para validar qualquer CNPJ com ou sem máscara, use validate_cnpj_qualquer.

validate_cnpj_qualquer

validate_cnpj_qualquer(cnpj)

Valida qualquer CNPJ: detecta se é numérico ou alfanumérico e delega.

CNPJs numéricos (somente dígitos, com ou sem máscara) usam validate_cnpj. CNPJs alfanuméricos (contêm letras A-Z, sem máscara) usam validate_cnpj_alfanumerico.

Tratamento de máscara (comportamento intencional e defensivo): - Este dispatcher remove apenas os separadores padrão de CNPJ: ponto, barra e traço. - Outros caracteres como espaços tornam o input inválido (retorna False). - Isso é diferente de normalizar_cnpj, que usa isalnum() e remove qualquer caractere não alfanumérico. A validação rejeita inputs com formato estranho antes de normalizar, mantendo fronteiras de entrada bem definidas. - Para normalizar antes de validar, use normalizar_cnpj() + validate_cnpj_alfanumerico().

normalizar_cnpj

normalizar_cnpj(cnpj)

Remove a máscara de um CNPJ (numérico ou alfanumérico) e normaliza para maiúsculas.

Para CNPJs numéricos: remove pontos, barra e traço. Para CNPJs alfanuméricos (IN RFB 2.229/2024): preserva as letras, remove máscara. Não valida o CNPJ - use validate_cnpj_qualquer antes se necessário.

validate_chave_nfe

validate_chave_nfe(chave)

Valida uma chave de acesso de NFe/NFCe (44 digitos).

A chave e composta por: cUF(2) + AAMM(4) + CNPJ(14) + mod(2) + serie(3) + nNF(9) + tpEmis(1) + cNF(8) + cDV(1) = 44 digitos.

Verifica apenas o digito verificador (modulo 11).

format_cpf

format_cpf(cpf, remover_mascara=False)

Formata um CPF com ou sem mascara.

Se remover_mascara=True, retorna apenas os 11 digitos numericos. Caso contrario, retorna no formato XXX.XXX.XXX-XX.

format_cnpj

format_cnpj(cnpj, remover_mascara=False)

Formata um CNPJ com ou sem mascara.

Se remover_mascara=True, retorna apenas os 14 digitos numericos. Caso contrario, retorna no formato XX.XXX.XXX/XXXX-XX.

format_chave_nfe

format_chave_nfe(chave)

Formata uma chave NFe em grupos de 4 digitos para legibilidade.

xml_utils

Utilitarios para parse e geracao de XML fiscal (NFe, NFSe, SPED).

parse_xml

parse_xml(content)

Parseia XML e retorna o elemento raiz.

Usa parser seguro contra XXE (resolve_entities=False, no_network=True). Lanca XMLParseError em caso de XML invalido.

xpath_text

xpath_text(element, xpath, namespaces=None)

Extrai o texto do primeiro elemento encontrado pelo XPath.

xpath_all_text

xpath_all_text(element, xpath, namespaces=None)

Extrai o texto de todos os elementos encontrados pelo XPath.

element_to_dict

element_to_dict(element)

Converte um elemento XML para dicionario Python.

Util para serializar partes de NFe para JSON.

strip_namespace

strip_namespace(xml_string)

Remove declaracoes de namespace do XML para facilitar parsing simples.

build_soap_envelope

build_soap_envelope(body_content, namespace='')

Monta um envelope SOAP para envio a webservices da SEFAZ.

Retorna a string XML completa com o envelope.

extract_soap_body

extract_soap_body(soap_response)

Extrai o conteudo do Body de uma resposta SOAP.

Lanca XMLParseError se o envelope for invalido.