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).

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_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.