peppol_check

Instructions

Implementation Reference

• mcp_einvoicing_de/tools/peppol_check.py:117-166 (handler)

  Async handler that validates input, performs Peppol DNS+SMP lookup via PeppolSMPClient, builds output model, and returns JSON result.

  async def handle_peppol_check(arguments: dict[str, Any]) -> list[types.TextContent]:
      """MCP handler for peppol_check."""
      try:
          params = PeppolCheckInput.model_validate(arguments)
      except Exception as exc:
          return [types.TextContent(type="text", text=json.dumps(format_error(str(exc))))]
  
      try:
          pid = PeppolParticipantId.parse(params.participant_id)
      except ValueError as exc:
          return [types.TextContent(type="text", text=json.dumps(format_error(str(exc))))]
  
      environment = (
          PeppolEnvironment.TEST
          if params.environment == "test"
          else PeppolEnvironment.PRODUCTION
      )
      client = PeppolSMPClient(environment=environment)
  
      try:
          lookup = await client.lookup_participant(pid)
      except EInvoicingError as exc:
          return [types.TextContent(type="text", text=json.dumps(format_error(str(exc))))]
  
      document_type_supported: bool | None = None
      access_point_url: str | None = None
      transport_profile: str | None = None
  
      if lookup.is_registered and params.document_type:
          document_type_supported = params.document_type in lookup.supported_document_types
          if document_type_supported:
              try:
                  service = await client.get_service_endpoint(
                      pid, params.document_type, smp_hostname=lookup.smp_hostname
                  )
                  access_point_url = service.endpoint_url
                  transport_profile = service.transport_profile
              except EInvoicingError as exc:
                  logger.warning("Service endpoint fetch failed: %s", exc)
  
      output = PeppolCheckOutput(
          is_registered=lookup.is_registered,
          participant_id=str(pid),
          document_type_supported=document_type_supported,
          access_point_url=access_point_url,
          transport_profile=transport_profile,
          lookup_details=lookup.to_dict(),
          error=lookup.error,
      )
      return [types.TextContent(type="text", text=output.model_dump_json(indent=2))]

• mcp_einvoicing_de/tools/peppol_check.py:45-67 (schema)

  PeppolCheckInput Pydantic model: participant_id (required), document_type (default BIS Billing 3.0), environment (production/test).

  class PeppolCheckInput(BaseModel):
      """Input schema for peppol_check."""
  
      participant_id: str = Field(
          ...,
          description=(
              "Peppol participant identifier in the format <scheme>:<value>. "
              "Examples: '0204:991-1234512345-06' (Leitweg-ID), "
              "'0088:4012345678901' (GLN), '9930:DE123456789' (VAT)."
          ),
      )
      document_type: str = Field(
          PEPPOL_BIS_BILLING_30,
          description=(
              "Peppol document type identifier to check capability for. "
              "Defaults to BIS Billing 3.0. "
              "[NEED: confirm DE PINT document type identifier once standardised]"
          ),
      )
      environment: str = Field(
          "production",
          description="Peppol environment: 'production' or 'test'.",
      )

• mcp_einvoicing_de/tools/peppol_check.py:70-83 (schema)

  PeppolCheckOutput Pydantic model: is_registered, participant_id, document_type_supported, access_point_url, transport_profile, lookup_details, error.

  class PeppolCheckOutput(BaseModel):
      """Output schema for peppol_check."""
  
      is_registered: bool
      participant_id: str
      document_type_supported: bool | None = Field(
          None, description="True if the participant supports the requested document type."
      )
      access_point_url: str | None = Field(
          None, description="AS4 endpoint URL of the participant's access point."
      )
      transport_profile: str | None = None
      lookup_details: dict[str, Any] = Field(default_factory=dict)
      error: str | None = None

• mcp_einvoicing_de/tools/peppol_check.py:86-114 (registration)

  TOOL_PEPPOL_CHECK: types.Tool definition with name='peppol_check', description, and input JSON schema.

  TOOL_PEPPOL_CHECK = types.Tool(
      name="peppol_check",
      description=(
          "Verify whether a German company is registered on the Peppol network "
          "and can receive electronic invoices via AS4. "
          "Performs a live DNS + SMP lookup using the participant's Peppol ID "
          "(Leitweg-ID, GLN, or VAT number). "
          "Returns registration status, supported document types, and AS4 endpoint URL."
      ),
      inputSchema={
          "type": "object",
          "required": ["participant_id"],
          "properties": {
              "participant_id": {
                  "type": "string",
                  "description": "Peppol participant ID, e.g. '0204:991-1234512345-06'.",
              },
              "document_type": {
                  "type": "string",
                  "description": "Peppol document type identifier (default: BIS Billing 3.0).",
              },
              "environment": {
                  "type": "string",
                  "enum": ["production", "test"],
                  "default": "production",
              },
          },
      },
  )

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

  Server imports TOOL_PEPPOL_CHECK and handle_peppol_check, adds to _ALL_TOOLS list and _TOOL_HANDLERS dict for dispatch.

  from mcp_einvoicing_de.tools.peppol_check import TOOL_PEPPOL_CHECK, handle_peppol_check
  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,
  }