# Copyright (C) 2023-2026 Sebastien Rousseau.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Model Context Protocol (MCP) server for Pacs008.
This server exposes the Pacs008 library's ISO 20022 ``pacs.008`` (FI-to-FI
Customer Credit Transfer) capabilities as MCP tools so that any MCP-compatible
client (Claude Desktop, IDEs, agents) can discover message types and scheme
profiles, validate records against the JSON Schema and against a rail's usage
guidelines, generate validated XML, validate raw XML against the bundled XSD,
and parse inbound ISO 20022 messages.
Every tool is a thin, typed wrapper over the ``pacs008`` library -- the same
package used by the CLI and REST API -- so all interfaces behave identically.
Tools return JSON-serializable data (dicts, lists, or strings); on a
:class:`ValueError` they return an ``{"error": ...}`` payload rather than
raising.
Launching the server:
* As a console script::
pacs008-mcp
* Programmatically::
from pacs008_mcp.server import main
main()
* In an MCP client config (e.g. Claude Desktop ``claude_desktop_config.json``)::
{
"mcpServers": {
"pacs008": {
"command": "pacs008-mcp"
}
}
}
stdio by default; ``--transport streamable-http`` or ``--transport sse``
listens on ``--host``/``--port`` instead. See :mod:`pacs008_mcp._cli`.
"""
import dataclasses
import json
import os
import shutil
import tempfile
from importlib.resources import files
from typing import Annotated, Any
from mcp.types import ToolAnnotations
from pacs008.constants import valid_xml_types
from pacs008.profiles import get_profile, list_profiles
from pacs008.standards.address import (
AddressPolicy,
PostalAddress,
Severity,
from_unstructured,
)
from pacs008.standards.address import (
validate_addresses as _lib_validate_addresses,
)
from pacs008.validation.bic_validator import validate_bic_format
from pacs008.validation.schema_validator import SchemaValidator
from pacs008.xml.generate_xml import generate_xml_string
from pacs008.xml.parser import parse
from pacs008.xml.validate_via_xsd import validate_xml_string_via_xsd
from pacs008_loader_mt103.loader import parse_mt103
from pydantic import Field
from pacs008_mcp import __version__, _cli
from pacs008_mcp._mcp_compat import build_server
# The shim picks FastMCP (mcp 1.x) or MCPServer (mcp 2.x) and reports
# the package version in serverInfo either way.
server = build_server("pacs008", __version__)
# Shared MCP tool annotations. Every tool in this server is a pure,
# side-effect-free reader over the pacs008 library: each tool computes solely
# from its arguments and the JSON Schemas / XSD templates bundled with the
# pacs008 library. None opens a caller-supplied filesystem path or reaches an
# external system, so all are marked ``readOnlyHint`` + ``idempotentHint``,
# never ``destructiveHint``, and closed-world (``openWorldHint=False``).
#
# These hints let MCP clients (and the Glama quality grader) reason about
# safety, caching, and auto-approval without executing the tool.
_PURE_READ = ToolAnnotations( # type: ignore[call-arg]
readOnlyHint=True,
destructiveHint=False,
idempotentHint=True,
openWorldHint=False,
)
# ``verify_bic_online`` is the one tool that MAY reach outside this process: if
# (and only if) a directory endpoint is configured it performs a read-only HTTP
# GET against that caller-supplied external system. It never mutates anything,
# so it stays ``readOnlyHint`` + ``idempotentHint``, but it is explicitly
# ``openWorldHint=True`` so MCP clients know it can contact an external service.
_ONLINE_READ = ToolAnnotations( # type: ignore[call-arg]
readOnlyHint=True,
destructiveHint=False,
idempotentHint=True,
openWorldHint=True,
)
# Environment variable that configures the BIC directory endpoint when no
# ``directory_url`` argument is supplied. There is no free authoritative SWIFT
# BIC directory API, so this server never ships or infers a BIC->institution
# mapping: institution details come solely from a directory the operator points
# it at (a commercial/central-bank/internal reference service).
_BIC_DIRECTORY_URL_ENV = "PACS008_BIC_DIRECTORY_URL"
# Human-readable names for each ISO 20022 message family, sourced verbatim from
# the pacs008 library's own generator docstrings (pacs008.xml.generate_xml).
_FAMILY_NAMES: dict[str, str] = {
"pacs.002": "FI to FI Payment Status Report",
"pacs.003": "FI to FI Customer Direct Debit",
"pacs.004": "Payment Return",
"pacs.007": "FI to FI Payment Reversal",
"pacs.008": "FI to FI Customer Credit Transfer",
"pacs.009": "Financial Institution Credit Transfer",
"pacs.010": "Financial Institution Direct Debit",
"pacs.028": "FI to FI Payment Status Request",
}
def _message_type_name(message_type: str) -> str:
"""Return the human name for a message type from its family prefix."""
family = ".".join(message_type.split(".")[:2])
return _FAMILY_NAMES.get(family, message_type)
# The pacs008 library's ``validate_path`` guard (invoked by
# ``generate_xml_string``) only permits paths under the current working
# directory or a system temp directory. The bundled templates live inside the
# installed package -- outside the CWD when the server is launched from
# anywhere else -- so we stage each message type's ``template.xml`` + ``.xsd``
# into a per-process temp directory (an allowed base) once and reuse them.
_STAGE_DIR = tempfile.mkdtemp(prefix="pacs008mcp-")
_STAGED: dict[str, tuple[str, str]] = {}
def _resolve_template_paths(message_type: str) -> tuple[str, str]:
"""Return staged (template.xml, xsd) paths for a message type.
The bundled files are resolved from the *installed* ``pacs008`` package via
:func:`importlib.resources.files` and copied into a temp directory the
library's path guard accepts, so the server works no matter what directory
it is launched from. Results are cached per message type.
"""
cached = _STAGED.get(message_type)
if cached is not None:
return cached
src = files("pacs008") / "templates" / message_type
dest = os.path.join(_STAGE_DIR, message_type)
os.makedirs(dest, exist_ok=True)
template_path = shutil.copy(
str(src / "template.xml"), os.path.join(dest, "template.xml")
)
xsd_path = shutil.copy(
str(src / f"{message_type}.xsd"),
os.path.join(dest, f"{message_type}.xsd"),
)
_STAGED[message_type] = (template_path, xsd_path)
return template_path, xsd_path
# ---------------------------------------------------------------------------
# Closed-set parameter enums.
#
# Every value below is derived from the pacs008 library's own source-of-truth
# constants / registry (never hardcoded here), so the accepted set stays in
# lockstep with the backend. The ``enum`` is JSON Schema metadata only -- it
# lets MCP clients constrain/auto-complete inputs -- while the library
# continues to enforce these values at runtime.
# ---------------------------------------------------------------------------
_MESSAGE_TYPE_VALUES: list[str] = sorted(valid_xml_types)
_MESSAGE_TYPE_LIST = ", ".join(f"'{v}'" for v in _MESSAGE_TYPE_VALUES)
_MessageType = Annotated[
str,
Field(
description=(
"A supported ISO 20022 pacs message type, e.g. 'pacs.008.001.08' "
"FI-to-FI Customer Credit Transfer. Must be exactly one of: "
f"{_MESSAGE_TYPE_LIST} (see list_message_types)."
),
json_schema_extra={"enum": _MESSAGE_TYPE_VALUES},
),
]
_SCHEME_VALUES: list[str] = sorted(list_profiles())
_SCHEME_LIST = ", ".join(f"'{v}'" for v in _SCHEME_VALUES)
_Scheme = Annotated[
str,
Field(
description=(
"A registered scheme / usage-guideline profile name "
"(case-insensitive), e.g. 'cbpr_plus', 'fedwire', 'chaps'. Must be "
f"one of: {_SCHEME_LIST} (see list_schemes)."
),
json_schema_extra={"enum": _SCHEME_VALUES},
),
]
# Address validation policy, derived from the pacs008 library's own
# ``AddressPolicy`` enum (source of truth). ``hybrid_or_structured`` is the
# November 14, 2026 cliff default: it rejects fully unstructured postal
# addresses across SWIFT CBPR+, HVPS+, T2 RTGS, CHAPS, Fedwire and Lynx.
_ADDRESS_POLICY_VALUES: list[str] = [policy.value for policy in AddressPolicy]
_ADDRESS_POLICY_LIST = ", ".join(f"'{v}'" for v in _ADDRESS_POLICY_VALUES)
_DEFAULT_ADDRESS_POLICY: str = AddressPolicy.HYBRID_OR_STRUCTURED.value
_AddressPolicy = Annotated[
str,
Field(
description=(
"Postal-address validation policy. 'unstructured_ok' permits any "
"form (pre-cutover / generic); 'hybrid_or_structured' rejects fully "
"unstructured addresses (the SWIFT CBPR+ UG2026 default; its "
"14 November 2026 start was deferred on 27 August 2026 and Swift "
"confirms new timing by December); 'structured_only' requires full "
f"structured form. Must be one of: {_ADDRESS_POLICY_LIST}."
),
json_schema_extra={"enum": _ADDRESS_POLICY_VALUES},
),
]
def _build_address(address: dict) -> PostalAddress:
"""Construct a ``PostalAddress`` from a snake_case field dict.
Raises ``ValueError``/``TypeError`` (e.g. bad country code, over-length
field, or an unknown key) which callers translate to an error payload.
"""
return PostalAddress(**address)
def _address_to_dict(address: PostalAddress) -> dict[str, Any]:
"""Serialize a ``PostalAddress`` to a JSON-friendly dict.
``adr_line`` (a tuple on the frozen dataclass) is emitted as a list.
"""
data = dataclasses.asdict(address)
data["adr_line"] = list(data["adr_line"])
return data
[docs]
def list_message_types() -> list[dict]:
"""List every supported ISO 20022 pacs message type and its human name.
Use this first, before any generation or validation call, to discover the
exact ``message_type`` strings this server accepts (e.g.
``pacs.008.001.08`` FI-to-FI Customer Credit Transfer). To learn a type's
required fields or full schema, call ``get_required_fields`` or
``get_input_schema`` instead.
Returns a list of ``{"message_type": ..., "name": ...}`` dictionaries, one
per supported message type.
"""
return [
{"message_type": mt, "name": _message_type_name(mt)}
for mt in _MESSAGE_TYPE_VALUES
]
[docs]
def list_schemes() -> list[dict]:
"""List every registered scheme / usage-guideline profile.
Scheme profiles (CBPR+, HVPS+, Fedwire, CHAPS, T2 RTGS, SCT Inst, generic)
layer rail-specific rules on top of base ISO 20022. Use this to discover
the ``scheme`` names accepted by ``get_scheme`` and ``validate_scheme``.
Registry aliases (e.g. ``cbpr+``, ``cbprplus``) collapse to their canonical
profile, so each profile appears exactly once. Returns a list of
``{"scheme": ..., "name": ...}`` dictionaries.
"""
canonical = {get_profile(name).name for name in _SCHEME_VALUES}
return [{"scheme": name, "name": name} for name in sorted(canonical)]
[docs]
def get_scheme(scheme: _Scheme) -> dict:
"""Return the rule attributes of a scheme / usage-guideline profile.
Use this to inspect a rail's constraints -- whether the UETR is mandatory,
the permitted charge bearers, remittance-info length cap, per-message
transaction cardinality, pinned message versions, and which parties must
carry an LEI -- before assembling or validating a batch.
Args:
scheme: A registered scheme profile name (see ``list_schemes``).
"""
try:
profile = get_profile(scheme)
return {
"scheme": scheme,
"name": profile.name,
"mr_version": profile.mr_version,
"uetr_required": profile.uetr_required,
"max_remit_info_len": profile.max_remit_info_len,
"allowed_charge_bearers": sorted(profile.allowed_charge_bearers),
"max_transactions_per_msg": profile.max_transactions_per_msg,
"lei_required_for": list(profile.lei_required_for()),
"pinned_versions": profile.pinned_versions(),
}
except ValueError as exc:
return {"error": str(exc)}
[docs]
def get_required_fields(
message_type: _MessageType,
) -> list[str]:
"""List only the required input field names for a pacs message type.
Use this for a quick checklist of the mandatory columns before building
payment records. For full type/format constraints (not just which fields
are required), call ``get_input_schema`` instead.
Args:
message_type: A supported ISO 20022 pacs message type.
"""
try:
return SchemaValidator(message_type).get_required_fields()
except ValueError as exc:
return [f"error: {exc}"]
[docs]
def validate_records(
message_type: _MessageType,
records: Annotated[
list[dict],
Field(
description=(
"One or more flat payment records, each a dict of field name "
"-> value; validated against the message type's input JSON "
"Schema (see get_input_schema / get_required_fields)."
)
),
],
) -> dict:
"""Validate flat payment records against a message type's JSON Schema.
Use this before ``generate_message`` to catch structural/type errors per
record and get a row-by-row error report. This checks JSON-Schema shape
only; to check a batch against a rail's usage guidelines use
``validate_scheme``.
Returns a report ``{"is_valid": bool, "total": int, "valid": int,
"errors": [...]}``.
Args:
message_type: A supported ISO 20022 pacs message type.
records: One or more flat payment records to validate.
"""
try:
validator = SchemaValidator(message_type)
except ValueError as exc:
return {"error": str(exc)}
total, valid, errors = validator.validate_batch(records)
report_errors = [
{
"row": row_idx,
"field": err.path,
"message": err.message,
"value": err.value,
}
for row_idx, row_errors in errors
for err in row_errors
]
return {
"is_valid": not report_errors,
"total": total,
"valid": valid,
"errors": report_errors,
}
[docs]
def validate_scheme(
scheme: _Scheme,
records: Annotated[
list[dict],
Field(
description=(
"One or more flat payment records, each a dict of field name "
"-> value; checked against the scheme's usage-guideline "
"business rules (charge bearer, UETR, remittance length, "
"per-message cardinality)."
)
),
],
) -> dict:
"""Validate payment records against a scheme's usage-guideline rules.
Use this to check a batch against a rail's rulebook (CBPR+, HVPS+,
Fedwire, CHAPS, T2 RTGS, SCT Inst) -- charge-bearer restrictions, UETR
presence, remittance-info length, and per-message transaction cardinality.
This is complementary to ``validate_records`` (JSON-Schema shape).
Returns ``{"scheme": str, "is_valid": bool, "total": int,
"violations": [...]}``.
Args:
scheme: A registered scheme profile name (see ``list_schemes``).
records: One or more flat payment records to check.
"""
try:
profile = get_profile(scheme)
except ValueError as exc:
return {"error": str(exc)}
violations = profile.validate_business_rules(records)
return {
"scheme": scheme,
"is_valid": not violations,
"total": len(records),
"violations": [dataclasses.asdict(v) for v in violations],
}
[docs]
def generate_message(
message_type: _MessageType,
records: Annotated[
list[dict],
Field(
description=(
"One or more flat payment records, each a dict of field name "
"-> value, from which the pacs XML is generated; run "
"validate_records first to surface record-level errors."
)
),
],
) -> str:
"""Generate a validated ISO 20022 pacs XML message from in-memory records.
This is the primary generation tool: pass payment records you already hold
in memory and receive an XSD-validated XML document; no file is written.
Run ``validate_records`` first to surface record-level errors, and
``list_message_types`` to confirm the ``message_type`` string.
Returns the validated XML document as a string, or an ``{"error": ...}``
payload (serialized) if generation fails.
Args:
message_type: A supported ISO 20022 pacs message type.
records: One or more flat payment records.
"""
if message_type not in valid_xml_types:
return json.dumps({"error": f"Invalid message type: {message_type}"})
try:
template_path, xsd_path = _resolve_template_paths(message_type)
return generate_xml_string(
records, message_type, template_path, xsd_path
)
except (ValueError, KeyError, RuntimeError) as exc:
return json.dumps({"error": str(exc)})
[docs]
def validate_xml(
message_type: _MessageType,
xml: Annotated[
str,
Field(
description=(
"A raw ISO 20022 XML document to validate against the bundled "
"XSD schema for the given message type."
)
),
],
) -> dict:
"""Validate a raw XML string against a message type's bundled XSD.
Use this to check an externally produced XML document against the official
ISO 20022 schema. To generate a document that is already XSD-validated,
use ``generate_message`` instead.
Returns ``{"message_type": str, "is_valid": bool}``, or ``{"error":
...}`` for an unsupported type or DTD entities; a bad year is invalid.
Args:
message_type: A supported ISO 20022 pacs message type.
xml: The raw XML document to validate.
"""
if message_type not in valid_xml_types:
return {"error": f"Invalid message type: {message_type}"}
_, xsd_path = _resolve_template_paths(message_type)
try: # DTD entities raise ValueError
is_valid = validate_xml_string_via_xsd(xml, xsd_path)
except ValueError as exc:
return {"error": f"refused XML: {exc!r}"}
return {"message_type": message_type, "is_valid": is_valid}
[docs]
def parse_message(
xml: Annotated[
str,
Field(
description=(
"A raw inbound ISO 20022 XML message (pacs.008 / pacs.002 / "
"pacs.004, optionally BAH-envelope-wrapped) to classify."
)
),
],
) -> dict:
"""Parse and classify an inbound ISO 20022 XML message.
Use this on the receiving side to identify what a message is -- its
``msg_def_idr`` (e.g. ``pacs.002.001.10``), family, version, and any
Business Application Header -- before processing it. Handles both bare
``Document`` messages and BAH-wrapped envelopes.
Returns a dict with ``msg_def_idr``, ``msg_family``, ``version``,
``root_local_name``, ``namespace_uri``, ``envelope_wrapped`` and ``bah``.
Args:
xml: The raw inbound XML message.
"""
try:
parsed = parse(xml)
except ValueError as exc:
return {"error": str(exc)}
bah = dataclasses.asdict(parsed.bah) if parsed.bah is not None else None
return {
"msg_def_idr": parsed.msg_def_idr,
"msg_family": parsed.msg_family,
"version": parsed.version,
"root_local_name": parsed.root_local_name,
"namespace_uri": parsed.namespace_uri,
"envelope_wrapped": parsed.envelope_wrapped,
"bah": bah,
}
[docs]
def convert_mt103(
mt103_text: Annotated[
str,
Field(
description=(
"A legacy SWIFT MT103 (single customer credit transfer) "
"payload as text. A raw '{4:...-}' block-4 envelope, trailing "
"whitespace and CRLF/LF differences are tolerated."
)
),
],
) -> dict:
"""Convert a legacy SWIFT MT103 into pacs.008-ready flat records.
This is the SWIFT MT-to-MX migration path (correspondent-banking MT103
coexistence with ISO 20022 ends November 2025): parse an MT103 text
payload and get back the flat pacs.008 record(s) that can be fed straight
into ``validate_records`` / ``generate_message``. An MT103 carries exactly
one transfer, so the ``records`` list always holds a single record. No
file is read or written.
Returns ``{"message_type": "pacs.008.001.08", "records": [{...}]}`` with
the parsed flat record, or an ``{"error": ...}`` payload if the MT103 is
missing a mandatory field (``:20:``, ``:32A:``, beneficiary) or malformed.
Args:
mt103_text: The MT103 payload as a string.
"""
try:
records = parse_mt103(mt103_text)
except ValueError as exc:
return {"error": str(exc)}
return {"message_type": "pacs.008.001.08", "records": records}
# ---------------------------------------------------------------------------
# November 2026 structured-address cliff tools.
#
# Fully unstructured postal addresses are decommissioned across SWIFT CBPR+,
# HVPS+, T2 RTGS, CHAPS, Fedwire and Lynx: any high-value or cross-border
# payment carrying an unstructured-only address is rejected at the rail. Swift
# deferred the CBPR+ start date on 27 August 2026 and confirms replacement
# timing by December; the other operators set their own and are unaffected.
# These tools wrap the pacs008 library's ``standards.address`` module
# so agents can classify, validate and repair addresses ahead of the cliff.
# ---------------------------------------------------------------------------
_ADDRESS_DICT_FIELD = Annotated[
dict,
Field(
description=(
"An ISO 20022 PostalAddress27 as a dict of snake_case fields, e.g. "
"{'strt_nm': 'High St', 'bldg_nb': '1', 'pst_cd': 'AB1 2CD', "
"'twn_nm': 'London', 'ctry': 'GB'} and optional 'adr_line' (list "
"of free-form lines). 'ctry' must be ISO 3166-1 alpha-2."
)
),
]
[docs]
def classify_address(address: _ADDRESS_DICT_FIELD) -> dict:
"""Classify a postal address as structured, hybrid, or unstructured.
Use this to see where an address stands against the SWIFT structured-address
rule: ``structured`` (town + country + structured detail, no free-form
lines), ``hybrid`` (town + country + 1-2 free-form ``adr_line`` lines, the
minimum CBPR+ UG2026 bar), or ``unstructured`` (free-form only — rejected
from the cliff date). To check acceptability under a policy use
``validate_address``; to upgrade legacy lines use ``repair_address``.
Returns ``{"classification": str, "is_structured": bool, "is_hybrid":
bool, "is_unstructured": bool, "has_structured_fields": bool}`` or an
``{"error": ...}`` payload.
Args:
address: The postal address as a dict of snake_case fields.
"""
try:
addr = _build_address(address)
except (ValueError, TypeError) as exc:
return {"error": str(exc)}
return {
"classification": addr.classify().value,
"is_structured": addr.is_structured(),
"is_hybrid": addr.is_hybrid(),
"is_unstructured": addr.is_unstructured(),
"has_structured_fields": addr.has_structured_fields,
}
[docs]
def validate_address(
address: _ADDRESS_DICT_FIELD,
policy: _AddressPolicy = _DEFAULT_ADDRESS_POLICY,
) -> dict:
"""Validate one postal address against an address policy.
Use this to decide whether an address will clear a rail. The default
``hybrid_or_structured`` policy is the November 14, 2026 cliff rule
(SWIFT CBPR+, HVPS+, T2 RTGS, CHAPS, Fedwire, Lynx): it rejects fully
unstructured addresses. Findings mirror the library's pipeline severity
(a policy rejection is a blocking finding).
Returns ``{"policy": str, "classification": str, "is_acceptable": bool,
"findings": [{"severity": str, "message": str}, ...]}`` or an
``{"error": ...}`` payload.
Args:
address: The postal address as a dict of snake_case fields.
policy: The validation policy to enforce (see the enum values).
"""
try:
resolved = AddressPolicy(policy)
except ValueError:
return {"error": f"Invalid policy: {policy!r}"}
try:
addr = _build_address(address)
except (ValueError, TypeError) as exc:
return {"error": str(exc)}
reason = addr.validate(resolved)
findings = (
[{"severity": Severity.BLOCK.value, "message": reason}]
if reason is not None
else []
)
return {
"policy": resolved.value,
"classification": addr.classify().value,
"is_acceptable": reason is None,
"findings": findings,
}
[docs]
def repair_address(
lines: Annotated[
list[str],
Field(
description=(
"Legacy unstructured address lines (free-form). Empty or "
"whitespace-only lines are skipped."
)
),
],
country: Annotated[
str,
Field(
description=(
"ISO 3166-1 alpha-2 country code (e.g. 'GB', 'US', 'DE', "
"'FR', 'JP') used to drive country-aware repair heuristics."
)
),
],
) -> dict:
"""Upgrade legacy unstructured address lines toward hybrid/structured form.
Experimental country-aware repair (``GB``, ``US``, ``DE``, ``FR``, ``JP``
have dedicated heuristics; other countries get a best-effort pass promoting
the last line to a town). Use this to lift pre-cliff data over the
November 14, 2026 bar; audit the output before submitting, and keep both
the original and derived address in your audit trail.
Returns ``{"address": {...}, "classification": str, "is_structured":
bool, "is_hybrid": bool}`` (so you can see the unstructured -> hybrid /
structured upgrade) or an ``{"error": ...}`` payload.
Args:
lines: Legacy unstructured address lines.
country: ISO 3166-1 alpha-2 country code driving the heuristics.
"""
try:
addr = from_unstructured(lines, country)
except ValueError as exc:
return {"error": str(exc)}
return {
"address": _address_to_dict(addr),
"classification": addr.classify().value,
"is_structured": addr.is_structured(),
"is_hybrid": addr.is_hybrid(),
}
[docs]
def validate_addresses(
addresses: Annotated[
list[dict],
Field(
description=(
"Payment-row dicts. The pipeline scans each row for columns of "
"the form '{party}_address_{field}' (party in debtor, creditor, "
"debtor_agent, creditor_agent, ultimate_debtor, "
"ultimate_creditor; field a snake_case PostalAddress field such "
"as twn_nm/ctry/strt_nm or adr_line_0..adr_line_6) and "
"validates each party's address."
)
),
],
policy: _AddressPolicy = _DEFAULT_ADDRESS_POLICY,
) -> dict:
"""Batch-validate every party address across a list of payment rows.
Use this before ``generate_message`` to catch addresses that will be
rejected at the rail. The default ``hybrid_or_structured`` policy enforces
the November 14, 2026 cliff. Each finding is reported per offending
``(row, party)`` pair.
Returns ``{"policy": str, "is_valid": bool, "total": int, "errors":
[{"row": int, "party": str, "severity": str, "message": str,
"classification": str}, ...]}`` or an ``{"error": ...}`` payload.
Args:
addresses: One or more payment-row dicts (see the field description).
policy: The validation policy to enforce (see the enum values).
"""
try:
resolved = AddressPolicy(policy)
except ValueError:
return {"error": f"Invalid policy: {policy!r}"}
try:
errors = _lib_validate_addresses(addresses, resolved)
except (ValueError, TypeError) as exc:
# A row whose address column breaks a PostalAddress rule (an
# over-length field, a bad country) fails at construction inside
# the library; report it like every other bad input.
return {"error": str(exc)}
return {
"policy": resolved.value,
"is_valid": not errors,
"total": len(addresses),
"errors": [
{
"row": err.row,
"party": err.party,
"severity": err.severity.value,
"message": err.message,
"classification": err.classification.value,
}
for err in errors
],
}
# ---------------------------------------------------------------------------
# BIC directory lookup.
#
# The pacs008 library ships an OFFLINE ISO 9362 structural validator
# (``validate_bic_format``): it checks the 8/11-char shape and that chars 5-6
# are a known ISO 3166-1 alpha-2 country code, but it knows nothing about which
# institution a BIC belongs to. There is no free authoritative SWIFT BIC
# directory API, so this tool never fabricates an institution name: it always
# does the offline structural check first, then -- only if the operator has
# configured a directory endpoint -- enriches the result with whatever the
# endpoint returns. With no endpoint configured it returns the structural
# result and says so, inventing nothing.
# ---------------------------------------------------------------------------
[docs]
def verify_bic_online(
bic: Annotated[
str,
Field(
description=(
"A BIC / SWIFT code to verify (ISO 9362), 8 or 11 characters, "
"e.g. 'DEUTDEFF' or 'DEUTDEFF500'. Spaces and hyphens are "
"stripped and the value is upper-cased before checking."
)
),
],
directory_url: Annotated[
str | None,
Field(
description=(
"Optional base URL of a BIC directory / reference-data endpoint "
"to enrich the result with institution details. The tool issues "
"a read-only HTTP GET with a '?bic=<BIC>' query parameter and "
"parses the JSON object it returns (name/city/country/status). "
"If omitted, the "
f"{_BIC_DIRECTORY_URL_ENV} environment variable is used; if "
"neither is set, only the offline structural result is returned "
"and NO institution name is inferred. No default/public "
"directory ships with this server."
)
),
] = None,
) -> dict:
"""Verify a BIC structurally, optionally enriched by a directory lookup.
Two-stage verification. First an OFFLINE ISO 9362 structural check (via the
pacs008 library): confirms the 8- or 11-character shape and that the
country code (chars 5-6) is a valid ISO 3166-1 alpha-2 code. A malformed
BIC returns ``{"bic": ..., "is_structurally_valid": False, "error": ...}``
and stops there.
Then, only if a directory endpoint is configured (the ``directory_url``
argument or the ``PACS008_BIC_DIRECTORY_URL`` environment variable), the
tool performs a read-only HTTP GET against that endpoint and folds whatever
institution ``name``/``city``/``country``/``status`` it returns into a
``directory`` sub-dict. There is no free authoritative SWIFT BIC directory,
so with NO endpoint configured the tool returns the structural result plus
a ``note`` and ``directory: null`` -- it never fabricates a bank name.
Online lookups require the optional ``online`` extra (``pip install
'pacs008-mcp[online]'``, which pulls in ``httpx``); without it the tool
still returns the structural result plus an ``error`` explaining the
missing extra. Endpoint 4xx/5xx responses and transport failures are
likewise reported as a graceful ``error`` alongside the structural result.
Returns a dict with ``bic``, ``is_structurally_valid``, ``bank_code``,
``country_code``, ``location_code``, ``branch_code``, ``length`` and either
``directory`` (a dict of institution details, or ``null``) plus, when
applicable, ``note`` / ``error``.
Args:
bic: The BIC / SWIFT code to verify.
directory_url: Optional directory endpoint for online enrichment.
"""
is_valid, error = validate_bic_format(bic)
if not is_valid:
return {
"bic": bic,
"is_structurally_valid": False,
"error": error,
}
bic_clean = bic.replace(" ", "").replace("-", "").upper()
result: dict[str, Any] = {
"bic": bic_clean,
"is_structurally_valid": True,
"bank_code": bic_clean[:4],
"country_code": bic_clean[4:6],
"location_code": bic_clean[6:8],
"branch_code": bic_clean[8:11] if len(bic_clean) == 11 else None,
"length": len(bic_clean),
}
endpoint = directory_url or os.environ.get(_BIC_DIRECTORY_URL_ENV)
if not endpoint:
result["directory"] = None
result["note"] = (
"No BIC directory endpoint configured; returning offline ISO 9362 "
"structural validation only. Pass a directory_url argument or set "
f"the {_BIC_DIRECTORY_URL_ENV} environment variable to enrich with "
"institution details. No institution name is inferred offline."
)
return result
try:
import httpx
except ImportError:
result["directory"] = None
result["error"] = (
"Online BIC lookup requires the optional 'online' extra. Install "
"it with: pip install 'pacs008-mcp[online]'."
)
return result
try:
response = httpx.get(endpoint, params={"bic": bic_clean}, timeout=10.0)
response.raise_for_status()
data = response.json()
except httpx.HTTPStatusError as exc:
result["directory"] = None
result["error"] = (
f"BIC directory returned HTTP {exc.response.status_code} for "
f"{bic_clean}."
)
return result
except httpx.HTTPError as exc:
result["directory"] = None
result["error"] = f"BIC directory request failed: {exc}"
return result
if not isinstance(data, dict):
result["directory"] = None
result["error"] = (
"BIC directory returned an unexpected (non-object) response."
)
return result
result["directory"] = {
"name": data.get("name"),
"city": data.get("city"),
"country": data.get("country"),
"status": data.get("status"),
}
return result
# MCP resources + prompt (the "Trinity" alongside the tools).
#
# Resources expose the same source-of-truth catalogues the reader tools return,
# so an agent can pin them as stable reference context without a tool call. The
# prompt teaches the recommended tool order for building/validating a pacs.008.
# Each function reuses the tool functions above verbatim -- no domain logic is
# reimplemented here.
# ---------------------------------------------------------------------------
[docs]
def message_types_resource() -> str:
"""Expose the supported pacs message types as a JSON resource.
Returns the same catalogue as the ``list_message_types`` tool -- each
supported ISO 20022 pacs ``message_type`` and its human name -- serialised
as JSON, so an agent can load the accepted type set as reference context
without a tool call. Mirrors ``list_message_types``, which takes no
arguments and cannot fail, so no error envelope is needed.
"""
return json.dumps(list_message_types())
[docs]
def schemes_resource() -> str:
"""Expose the registered scheme / usage-guideline profiles as JSON.
Returns the same catalogue as the ``list_schemes`` tool -- each canonical
rail profile (CBPR+, HVPS+, Fedwire, CHAPS, T2 RTGS, SCT Inst, generic) --
serialised as JSON, so an agent can load the accepted scheme set as
reference context without a tool call. Mirrors ``list_schemes``, which
takes no arguments and cannot fail, so no error envelope is needed.
"""
return json.dumps(list_schemes())
[docs]
def scheme_resource(
scheme_id: Annotated[
str,
Field(
description=(
"A registered scheme / usage-guideline profile name "
"(case-insensitive), e.g. 'cbpr_plus', 'fedwire', 'chaps' "
"(see list_schemes / the pacs008://schemes resource)."
)
),
],
) -> str:
"""Expose one scheme profile's rule attributes as a templated JSON resource.
A templated MCP Resource returning the same payload as the ``get_scheme``
tool for a single scheme: the UETR requirement, permitted charge bearers,
remittance-info length cap, per-message transaction cardinality, pinned
message versions, and LEI-required parties. Lets an agent pin
``pacs008://scheme/cbpr_plus`` as stable reference context for the rail it
is building against. An unknown ``scheme_id`` yields the same
``{"error": ...}`` payload as ``get_scheme`` (serialised), so no separate
error handling is added here.
Args:
scheme_id: A registered scheme profile name (see ``list_schemes``).
"""
return json.dumps(get_scheme(scheme_id))
[docs]
def build_pacs008_message(
goal: Annotated[
str,
Field(
description=(
"The payments task in plain language, e.g. 'send a EUR "
"customer credit transfer over CBPR+' or 'migrate this MT103 "
"to pacs.008'."
)
),
] = "",
) -> str:
"""Guided prompt teaching the recommended tool order for a pacs.008.
The MCP client sends this to the model to establish the correct
build-and-validate sequence so it uses the server's tools in order instead
of guessing a message type or skipping a validation gate.
Args:
goal: The payments task in plain language.
Returns:
A prompt string instructing the model how to proceed.
"""
goal_line = (
f' The task is: "{goal.strip()}".' if goal and goal.strip() else ""
)
return (
f"You have the pacs008 ISO 20022 server.{goal_line} Work in this "
"order: call list_message_types() to confirm the exact message_type "
"(or read the pacs008://message-types resource). If you are starting "
"from a legacy SWIFT MT103, call convert_mt103(mt103_text) to get the "
"flat pacs.008 record(s). Call get_required_fields(message_type) and "
"get_input_schema(message_type) to learn the fields, then assemble one "
"flat record per instruction. Check structural shape with "
"validate_records(message_type, records); check the rail's rulebook "
"with validate_scheme(scheme, records) after inspecting get_scheme("
"scheme) (or the pacs008://scheme/{scheme_id} resource); and check "
"party addresses against the CBPR+ structured-address rule "
"with validate_addresses(addresses). Only once all checks pass, call "
"generate_message(message_type, records) for the XSD-validated XML. To "
"verify an externally produced document use validate_xml(message_type, "
"xml); to classify an inbound message use parse_message(xml)."
)
# Tools, resources and the prompt are registered here, in definition
# order, rather than with decorators on each function: mutmut 3 never
# mutates a decorated function, so the decorator form left every
# handler outside mutation testing. The registered object is the same
# function, docstring and signature, and clients list them in this
# order.
server.tool(title="List pacs message types", annotations=_PURE_READ)(
list_message_types
)
server.tool(title="List scheme profiles", annotations=_PURE_READ)(list_schemes)
server.tool(title="Get scheme profile rules", annotations=_PURE_READ)(
get_scheme
)
server.tool(title="Get required fields", annotations=_PURE_READ)(
get_required_fields
)
server.tool(title="Get input JSON Schema", annotations=_PURE_READ)(
get_input_schema
)
server.tool(title="Validate records against schema", annotations=_PURE_READ)(
validate_records
)
server.tool(title="Validate records against a scheme", annotations=_PURE_READ)(
validate_scheme
)
server.tool(title="Generate pacs XML from records", annotations=_PURE_READ)(
generate_message
)
server.tool(title="Validate XML against XSD", annotations=_PURE_READ)(
validate_xml
)
server.tool(title="Parse inbound ISO 20022 XML", annotations=_PURE_READ)(
parse_message
)
server.tool(title="Convert MT103 to pacs.008 records", annotations=_PURE_READ)(
convert_mt103
)
server.tool(title="Classify a postal address", annotations=_PURE_READ)(
classify_address
)
server.tool(title="Validate a postal address", annotations=_PURE_READ)(
validate_address
)
server.tool(title="Repair an unstructured address", annotations=_PURE_READ)(
repair_address
)
server.tool(
title="Validate addresses in payment rows", annotations=_PURE_READ
)(validate_addresses)
server.tool(
title="Verify a BIC (structural + optional directory lookup)",
annotations=_ONLINE_READ,
)(verify_bic_online)
server.resource(
"pacs008://message-types", title="pacs message-type catalogue"
)(message_types_resource)
server.resource("pacs008://schemes", title="Scheme profile catalogue")(
schemes_resource
)
server.resource("pacs008://scheme/{scheme_id}", title="Scheme profile rules")(
scheme_resource
)
server.prompt(title="Build a pacs.008 message")(build_pacs008_message)
[docs]
def main(argv: list[str] | None = None) -> None:
"""Run the pacs008 MCP server (the ``pacs008-mcp`` entry point).
stdio by default; ``--transport streamable-http`` or ``--transport sse``
listens on ``--host``/``--port`` instead. See :mod:`pacs008_mcp._cli`.
Args:
argv: Command-line arguments; ``None`` reads ``sys.argv[1:]``.
"""
_cli.serve(server, argv, "pacs008-mcp", __version__)
if __name__ == "__main__":
main()