tax_rules

Instructions

Implementation Reference

• mcp_einvoicing_de/tools/tax_rules.py:192-236 (handler)

  The main MCP handler function for the tax_rules tool. Accepts a query and optional context, then looks up German VAT rates, reverse charge cases, exemptions, and VATEX codes based on keyword matching. Returns results as JSON text content.

  async def handle_tax_rules(arguments: dict[str, Any]) -> list[types.TextContent]:
      """MCP handler for tax_rules."""
      try:
          params = TaxRulesInput.model_validate(arguments)
      except Exception as exc:
          return [types.TextContent(type="text", text=json.dumps(format_error(str(exc))))]
  
      query_lower = params.query.lower()
      results: list[dict[str, Any]] = []
      notes: list[str] = []
  
      if any(kw in query_lower for kw in ("rate", "satz", "19", "7", "prozent")):
          results.extend(_GERMAN_VAT_RATES.values())
  
      if any(kw in query_lower for kw in ("reverse", "13b", "§13b", "steuerschuld", "umkehr")):
          results.extend(_REVERSE_CHARGE_CASES)
          if params.context:
              context_lower = params.context.lower()
              results = [
                  r for r in results
                  if any(kw in r.get("description_en", "").lower() for kw in context_lower.split())
              ] or results  # Fall back to all if filter yields nothing
  
      if any(kw in query_lower for kw in ("exempt", "befreit", "§4", "zero", "null", "klein", "19 ustg")):
          results.extend(_EXEMPTIONS)
  
      if any(kw in query_lower for kw in ("vatex", "reason code", "exemption code")):
          vatex_codes = [
              {"vatex_code": r.get("vatex_code"), "description_en": r.get("description_en"), "paragraph": r.get("paragraph")}
              for r in _EXEMPTIONS + _REVERSE_CHARGE_CASES
              if r.get("vatex_code")
          ]
          results.extend(vatex_codes)
  
      if not results:
          notes.append(
              "No specific rules matched the query. "
              "Try: 'rates', 'reverse_charge', 'exemptions', 'vatex_codes', or '13b'."
          )
          results = list(_GERMAN_VAT_RATES.values())
  
      notes.append("Data reflects German VAT law as of 2025-01-01. [NEED: implement rule versioning]")
  
      output = TaxRulesOutput(query=params.query, results=results, notes=notes)
      return [types.TextContent(type="text", text=output.model_dump_json(indent=2))]

• mcp_einvoicing_de/tools/tax_rules.py:131-148 (schema)

  Input schema (TaxRulesInput) with required 'query' string and optional 'context' string for filtering results.

  class TaxRulesInput(BaseModel):
      """Input schema for tax_rules."""
  
      query: str = Field(
          ...,
          description=(
              "What to look up. Examples: 'reverse_charge', 'rates', 'exemptions', "
              "'kleinunternehmer', '13b', 'zero_rate', 'vatex_codes', "
              "or a free-text question about German VAT."
          ),
      )
      context: str | None = Field(
          None,
          description=(
              "Optional context about the transaction, e.g. 'construction services' "
              "or 'intra-community supply'. Used to filter relevant rules."
          ),
      )

• mcp_einvoicing_de/tools/tax_rules.py:151-163 (schema)

  Output schema (TaxRulesOutput) containing the query, results list, notes, and a legal disclaimer.

  class TaxRulesOutput(BaseModel):
      """Output schema for tax_rules."""
  
      query: str
      results: list[dict[str, Any]]
      notes: list[str] = Field(default_factory=list)
      legal_disclaimer: str = Field(
          default=(
              "This information is provided for technical reference only and does not "
              "constitute legal or tax advice. Always consult a qualified tax adviser "
              "(Steuerberater) for binding guidance. Rules are subject to legislative changes."
          )
      )

• mcp_einvoicing_de/tools/tax_rules.py:166-189 (registration)

  TOOL_TAX_RULES constant — the MCP Tool definition with name='tax_rules', description, and input JSON schema.

  TOOL_TAX_RULES = types.Tool(
      name="tax_rules",
      description=(
          "Query German VAT rules for e-invoicing. "
          "Returns structured information about VAT rates (19%, 7%), VAT category codes, "
          "reverse charge rules under §13b UStG, zero-rate and exemption provisions (§4 UStG), "
          "intra-community supply rules, and VATEX exemption reason codes. "
          "For use when building invoice creation logic or validating VAT treatment."
      ),
      inputSchema={
          "type": "object",
          "required": ["query"],
          "properties": {
              "query": {
                  "type": "string",
                  "description": "What to look up: 'rates', 'reverse_charge', 'exemptions', 'vatex_codes', etc.",
              },
              "context": {
                  "type": "string",
                  "description": "Optional transaction context to filter results.",
              },
          },
      },
  )

• mcp_einvoicing_de/server.py:21-43 (registration)

  Registration of TOOL_TAX_RULES in the _ALL_TOOLS list and the handle_tax_rules function in the _TOOL_HANDLERS dict, plus the import from the tax_rules module.

  from mcp_einvoicing_de.tools.tax_rules import TOOL_TAX_RULES, handle_tax_rules
  
  LOG_LEVEL = os.environ.get("EINVOICING_DE_LOG_LEVEL", "INFO").upper()
  logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO))
  logger = logging.getLogger(__name__)
  
  _ALL_TOOLS: list[types.Tool] = [
      TOOL_INVOICE_CREATE,
      TOOL_INVOICE_VALIDATE,
      TOOL_INVOICE_PARSE,
      TOOL_INVOICE_CONVERT,
      TOOL_PEPPOL_CHECK,
      TOOL_TAX_RULES,
  ]
  
  _TOOL_HANDLERS: dict[str, Any] = {
      "invoice_create": handle_invoice_create,
      "invoice_validate": handle_invoice_validate,
      "invoice_parse": handle_invoice_parse,
      "invoice_convert": handle_invoice_convert,
      "peppol_check": handle_peppol_check,
      "tax_rules": handle_tax_rules,
  }