Pular para o conteúdo
API de emissão de notas do FazNota

Exemplos em Python

Setup

Terminal window
pip install requests python-dotenv
.env
MYSE_API_BASE=https://api.mysebr.com.br/nfemyse-v3/rest
MYSE_API_TOKEN=seu_token_aqui

Cliente HTTP centralizado

myse_api.py
import os
import time
from dataclasses import dataclass
from typing import Any, Optional
import requests
from dotenv import load_dotenv


load_dotenv()


BASE = os.environ["MYSE_API_BASE"]
TOKEN = os.environ["MYSE_API_TOKEN"]


@dataclass
class ApiError(Exception):
    http_status: int
    message: str
    payload: Any
    request_id: Optional[str] = None


    def __str__(self):
        return f"[{self.http_status}] {self.message} (request_id={self.request_id})"


def m_api(method: str, path: str, body: dict | None = None, timeout: int = 30) -> dict:
    headers = {
        "Authorization": f"Token {TOKEN}",
        "Content-Type": "application/json",
    }
    res = requests.request(
        method, f"{BASE}{path}", json=body, headers=headers, timeout=timeout
    )


    request_id = res.headers.get("X-Request-Id")


    # Erros HTTP duros
    if not res.ok:
        try:
            payload = res.json()
            msg = payload.get("error", {}).get("message", f"HTTP {res.status_code}")
        except Exception:
            payload = {}
            msg = f"HTTP {res.status_code}"
        raise ApiError(res.status_code, msg, payload, request_id)


    json = res.json()


    # Erro de domínio
    if json.get("status") == "999":
        raise ApiError(
            200,
            json.get("data", {}).get("erro") or json.get("descricao", "erro"),
            json,
            (json.get("meta") or {}).get("request_id") or request_id,
        )


    return json

Cenário: emitir NFCe com polling

import time


def emitir_nfce(pedido: dict) -> dict:
    # 1. Emissão
    body = {
        "serie": "65",
        "cfop": "5102",
        "numero-origem": f"PED-{pedido['id']}",
        "tp_pagamento": pedido["forma_pagamento"],
        "cliente": pedido["cpf_cnpj_cliente"],
        "itens": [
            {
                "produto": i["sku"],
                "quantidade": str(i["qtd"]),
                "valor-unitario": str(i["preco"]),
            }
            for i in pedido["itens"]
        ],
    }


    emissao = m_api("POST", "/nfce/emissao", body)
    if emissao["status"] not in ("001", "050"):
        raise RuntimeError(f"Emissão falhou: {emissao['descricao']}")


    recibo = emissao["data"]["recibo"]
    print(f"[emissao] recibo={recibo} request_id={emissao['meta']['request_id']}")


    # 2. Polling
    TERMINAIS = {"004", "010", "900", "999"}
    wait = 5.0
    inicio = time.time()


    while time.time() - inicio < 5 * 60:
        time.sleep(wait)
        consulta = m_api("GET", f"/nfce/cupom/{recibo}")
        print(f"[polling] status={consulta['status']}")


        if consulta["status"] in TERMINAIS:
            if consulta["status"] == "004":
                return consulta["data"]
            raise RuntimeError(f"Status terminal não-sucesso: {consulta['status']}")


        wait = min(wait * 1.5, 30)


    raise TimeoutError(f"Timeout aguardando recibo {recibo}")

Retry em erros transitórios

def com_retry(fn, max_tentativas=3):
    for i in range(1, max_tentativas + 1):
        try:
            return fn()
        except ApiError as e:
            if e.payload and e.payload.get("status") == "999":
                raise  # Validação: não retentar
            if i == max_tentativas:
                raise
            time.sleep(2 ** (i - 1))  # 1s, 2s, 4s
        except (requests.Timeout, requests.ConnectionError):
            if i == max_tentativas:
                raise
            time.sleep(2 ** (i - 1))


# Uso
nfce = com_retry(lambda: emitir_nfce(pedido))

Helper de polling reutilizável

TERMINAIS = {"004", "010", "900", "999"}


def aguardar_recibo(path_base: str, recibo: str, timeout: int = 300) -> dict:
    """Polling com backoff exponencial até status terminal."""
    wait = 5.0
    inicio = time.time()


    while time.time() - inicio < timeout:
        time.sleep(wait)
        resp = m_api("GET", f"{path_base}/{recibo}")
        if resp["status"] in TERMINAIS:
            return resp
        wait = min(wait * 1.5, 30)


    raise TimeoutError(f"Timeout aguardando recibo {recibo}")

Emitir NF-e (com transporte completo)

def emitir_nfe(pedido: dict) -> dict:
    body = {
        "serie": "1",
        "cfop": "5102",
        "numero-origem": f"PED-{pedido['id']}",
        "tipo": "S",
        "finalidade": "N",
        "valor-despesas": "0.00",
        "cliente": pedido["cpf_cnpj_cliente"],
        "transporte": {
            "modalidade": "0",
            "transportadora": pedido["transportadora"],
            "placa-veiculo": pedido["placa"],
            "valor-frete": pedido.get("frete", "0.00"),
            "valor-seguro": "0.00",
            "peso-bruto": pedido["peso_bruto"],
            "peso-liquido": pedido["peso_liquido"],
            "quantidade-volume": "1",
            "especie-volume": "Caixa",
            "marca-volume": "Embalagem",
        },
        "itens": [
            {
                "produto": i["sku"],
                "quantidade": str(i["qtd"]),
                "valor-unitario": str(i["preco"]),
            }
            for i in pedido["itens"]
        ],
        "faturas": [
            {"valor": str(f["valor"]), "data": f["data"]}
            for f in pedido.get("faturas", [])
        ],
    }


    emissao = m_api("POST", "/nfe/emissao", body)
    if emissao["status"] not in ("001", "050"):
        raise RuntimeError(f"Emissão falhou: {emissao['descricao']}")


    resultado = aguardar_recibo("/nfe/nota", emissao["data"]["recibo"])
    if resultado["status"] != "004":
        raise RuntimeError(f"Status terminal não-sucesso: {resultado['status']}")
    return resultado["data"]  # inclui sefaz.chave, url-danfe, url-xml

Cancelar NF-e

def cancelar_nfe(chave: str, motivo: str) -> dict:
    assert 15 <= len(motivo) <= 255, "motivo deve ter entre 15 e 255 caracteres"
    cancel = m_api("POST", "/nfe/cancelamento", {"chave": chave, "motivo": motivo})
    if cancel["status"] != "001":
        raise RuntimeError(cancel["descricao"])
    return aguardar_recibo("/nfe/cancelamento", cancel["data"]["recibo"])

Lançar carta de correção (CC-e)

def carta_correcao_nfe(chave: str, texto_correcao: str) -> dict:
    """O campo 'motivo' na CC-e é o TEXTO da correção (xCorrecao)."""
    assert 15 <= len(texto_correcao) <= 255, "texto deve ter 15-255 chars"
    cce = m_api("POST", "/nfe/correcao", {"chave": chave, "motivo": texto_correcao})
    if cce["status"] != "001":
        raise RuntimeError(cce["descricao"])
    return aguardar_recibo("/nfe/correcao", cce["data"]["recibo"])

Cancelar NFC-e

def cancelar_nfce(chave: str, motivo: str) -> dict:
    assert 15 <= len(motivo) <= 255, "motivo deve ter 15-255 chars"
    cancel = m_api("POST", "/nfce/cancelamento", {"chave": chave, "motivo": motivo})
    if cancel["status"] != "001":
        raise RuntimeError(cancel["descricao"])
    return aguardar_recibo("/nfce/cancelamento", cancel["data"]["recibo"])

Emitir NFS-e (Nota Fiscal de Serviço)

def emitir_nfse(servico: dict) -> dict:
    body = {
        "numero-origem": f"OS-{servico['id']}",
        "observacao": servico["observacao"],
        "nome-contato": servico.get("nome_contato"),
        "telefone-contato": servico.get("telefone_contato"),
        "data-fim": servico["data_fim"],
        "cliente": servico["cpf_cnpj_cliente"],
        "itens": [
            {
                "produto": i["codigo_servico"],
                "quantidade": str(i["qtd"]),
                "valor-unitario": str(i["preco"]),
            }
            for i in servico["itens"]
        ],
    }


    emissao = m_api("POST", "/nfse/emissao", body)
    if emissao["status"] not in ("001", "050"):
        raise RuntimeError(f"Emissão NFSe falhou: {emissao['descricao']}")


    resultado = aguardar_recibo("/nfse/nota", emissao["data"]["recibo"])
    if resultado["status"] != "004":
        raise RuntimeError(f"Status terminal não-sucesso: {resultado['status']}")
    return resultado["data"]

Cancelar NFS-e

def cancelar_nfse(chave: str, motivo: str) -> dict:
    assert 15 <= len(motivo) <= 255, "motivo deve ter 15-255 chars"
    return m_api("POST", "/nfse/cancelamento", {"chave": chave, "motivo": motivo})

Reconhecer nota de fornecedor (MD-e)

CODIGOS_RECONHECIMENTO = {"210200", "210210", "210220", "210240"}
# 210200 — Confirmação da operação (terminal, irreversível)
# 210210 — Ciência da operação (provisório, 15 dias)
# 210220 — Desconhecimento da operação
# 210240 — Operação não realizada


def reconhecer_nota_fornecedor(chave: str, codigo: str) -> dict:
    if codigo not in CODIGOS_RECONHECIMENTO:
        raise ValueError(f"Código inválido: {codigo}")


    resp = m_api("POST", "/nffornecedor/reconhecimento", {
        "chave": chave,
        "codigo-reconhecimento": codigo,
    })
    if resp["status"] != "001":
        raise RuntimeError(resp["descricao"])
    return aguardar_recibo("/nffornecedor", resp["data"]["recibo"])

Listar com paginação

def listar_todos_clientes() -> list:
    todos = []
    page = 1
    page_size = 100


    while True:
        resp = m_api("GET", f"/clientes?page={page}&page_size={page_size}")
        if resp["status"] != "005" or not isinstance(resp.get("data"), list):
            break
        todos.extend(resp["data"])
        if len(resp["data"]) < page_size:
            break
        page += 1
        time.sleep(0.2)


    return todos

Tratamento de erros

try:
    nfce = emitir_nfce(pedido)
    print("NFCe emitida:", nfce.get("sefaz", {}).get("chave"))
except ApiError as e:
    print(f"API Error: {e}")
    # Log para suporte com request_id
    logger.error(
        "API call failed",
        extra={
            "request_id": e.request_id,
            "http_status": e.http_status,
            "payload": e.payload,
        },
    )
except TimeoutError as e:
    print(f"Timeout: {e}")

Variação: usando httpx (async)

import httpx


async def m_api_async(method, path, body=None):
    async with httpx.AsyncClient(timeout=30) as client:
        res = await client.request(
            method,
            f"{BASE}{path}",
            json=body,
            headers={"Authorization": f"Token {TOKEN}"},
        )
        # ... mesma lógica de tratamento

Health check (sem autenticação)

import requests


def health() -> dict:
    res = requests.get(f"{BASE}/health", timeout=5)
    return res.json()
    # {"status": "ok", "database": "ok", "timestamp": "..."}

Alterar nota antes da emissão (PUT)

Disponível em NF-e (/nfe/{id}), NFC-e (/nfce/{id}) e NFS-e (/nfse/{id}). Só funciona enquanto a nota ainda não foi submetida para emissão.

def alterar_nota(tipo: str, id_nota: str, dados: dict) -> dict:
    """tipo: 'nfe' | 'nfce' | 'nfse'"""
    return m_api("PUT", f"/{tipo}/{id_nota}", dados)


# Uso:
alterar_nota("nfe", "12345", {
    "serie": "1",
    "cfop": "5102",
    "cliente": "12345678000199",
    "itens": [{"produto": "SKU-001", "quantidade": "3", "valor-unitario": "89.90"}],
})

Consulta detalhada por recibo

nfe_completa = m_api("GET", f"/nfe/nota/{recibo}/completa")
nfce_completa = m_api("GET", f"/nfce/cupom/{recibo}/completa")

Listagens

# NF-es paginadas
nfes = m_api("GET", "/nfe?page=1&page_size=20")


# NFC-es
nfces = m_api("GET", "/nfce?page=1&page_size=50")


# NFS-es com filtro de data
nfses = m_api("GET", "/nfse?dataInicial=2026-05-01&page=1&page_size=20")


# Notas de fornecedor
fornecedor = m_api("GET", "/nffornecedor?page=1&page_size=20")


# Forma legada (path-based)
nfes_legado = m_api("GET", "/nfe/paginacao/1/20")

Clientes (CRUD completo)

def cadastrar_pf(dados: dict) -> dict:
    return m_api("POST", "/clientes", {
        "tipo": "F",
        "nome": dados["nome"],
        "cpf": dados["cpf"],
        "rg": dados.get("rg"),
        "consumidor-final": "1",
        "contato": {
            "email": dados.get("email"),
            "telefone": dados.get("telefone"),
        },
        "endereco": dados["endereco"],
    })


def cadastrar_pj(dados: dict) -> dict:
    return m_api("POST", "/clientes", {
        "tipo": "J",
        "razao-social": dados["razao_social"],
        "nome-fantasia": dados.get("nome_fantasia"),
        "cnpj": dados["cnpj"],
        "inscricao-municipal": dados.get("inscricao_municipal"),
        "tipo-inscricao-estadual": dados.get("contribuinte", "9"),
        "inscricao-estadual": dados.get("inscricao_estadual", "ISENTO"),
        "consumidor-final": dados.get("consumidor_final", "0"),
        "contato": {
            "email": dados.get("email"),
            "telefone": dados.get("telefone"),
        },
        "endereco": dados["endereco"],
    })


# Consultar por CPF/CNPJ
cliente = m_api("GET", "/clientes/12345678000199")


# Buscar por nome
encontrados = m_api("GET", "/clientes/busca/exemplo")
top = m_api("GET", "/clientes/busca/top_clientes")


# Atualizar
m_api("PUT", "/clientes/12345678000199", {
    "tipo": "J",
    "razao-social": "Empresa Exemplo (Novo Nome) Ltda",
})


# Excluir
m_api("DELETE", "/clientes/12345678000199")

Produtos (CRUD completo)

def cadastrar_produto(produto: dict) -> dict:
    return m_api("POST", "/produtos", {
        "referencia": produto["sku"],
        "nome": produto["nome"],
        "valor": str(produto["valor"]),
        "medida": produto.get("medida", "UN"),
        "categoria": produto.get("categoria"),
        "ncm": produto["ncm"],
        "origem": produto.get("origem", "0"),
        "cfop-preferencial": produto.get("cfop", "5102"),
        "tipo-produto": "P",
        "configuracoes-fiscais": produto.get("config_fiscal", []),
    })


def cadastrar_servico(servico: dict) -> dict:
    return m_api("POST", "/produtos", {
        "referencia": servico["sku"],
        "nome": servico["nome"],
        "valor": str(servico["valor"]),
        "medida": "H",
        "tipo-produto": "S",
        "codigo-servico": servico["codigo_servico"],
    })


# Consultar por referência
produto = m_api("GET", "/produtos/SKU-001")


# Buscar por nome (1º resultado)
por_nome = m_api("GET", "/produtos/nome/Camiseta")


# Listar paginado com busca (use _todos para não filtrar)
todos = m_api("GET", "/produtos/paginacao/1/20/_todos")
filtrados = m_api("GET", "/produtos/paginacao/1/20/Camiseta")


# Atualizar
m_api("PUT", "/produtos/SKU-001", {
    "referencia": "SKU-001",
    "nome": "Camiseta Polo M (Novo)",
    "valor": "99.90",
})


# Excluir
m_api("DELETE", "/produtos/SKU-001")