# 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 Acmt001.
This server exposes the Acmt001 library's ISO 20022 ``acmt`` capabilities as
MCP tools so that any MCP-compatible client (Claude Desktop, IDEs, agents) can
discover message types, inspect input schemas, validate records and financial
identifiers, and generate validated XML messages.
Every tool is a thin, typed wrapper over :mod:`acmt001.services` -- the single
shared facade also used by the CLI, REST API, and LSP server -- so all
interfaces behave identically. Tools return JSON-serializable data (dicts,
lists, or strings); on a :class:`ValueError` they return an ``{"error": ...}``
dictionary rather than raising.
Launching the server:
* As a console script (installed with the ``servers`` extra)::
acmt001-mcp
* Programmatically::
from acmt001_mcp.server import main
main()
* In an MCP client config (e.g. Claude Desktop ``claude_desktop_config.json``)::
{
"mcpServers": {
"acmt001": {
"command": "acmt001-mcp"
}
}
}
stdio by default; ``--transport streamable-http`` or ``--transport sse``
listens on ``--host``/``--port`` instead. See :mod:`acmt001_mcp._cli`.
"""
import json
import re
import time
from typing import Annotated
from acmt001 import services
from acmt001.constants import valid_xml_types
from mcp.types import ToolAnnotations
from pydantic import Field
from acmt001_mcp import __version__, _cli
from acmt001_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("acmt001", __version__)
# Shared MCP tool annotations. Every tool in this server is a pure,
# side-effect-free reader over the acmt001 ``services`` facade: each tool
# computes solely from its arguments and the JSON Schemas / XSD templates
# bundled with the acmt001 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,
)
# Annotations for the single tool that reaches an external system. Unlike every
# other tool here, ``verify_lei_online`` performs a live HTTP GET against the
# public GLEIF LEI register, so it is open-world (``openWorldHint=True``). It is
# still a non-destructive, idempotent read: it never mutates remote state, and
# the same LEI yields the same record barring an upstream data change.
_EXTERNAL = ToolAnnotations( # type: ignore[call-arg]
readOnlyHint=True,
destructiveHint=False,
idempotentHint=True,
openWorldHint=True,
)
# GLEIF (Global Legal Entity Identifier Foundation) public REST API endpoint for
# a single LEI record. See https://api.gleif.org/. This is the one external
# system this server contacts, and only from ``verify_lei_online``.
_GLEIF_LEI_RECORD_URL = "https://api.gleif.org/api/v1/lei-records/{lei}"
# ---------------------------------------------------------------------------
# Closed-set parameter enums.
#
# Every value below is derived from the acmt001 library's own source-of-truth
# constants (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 ``services`` facade 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 acmt message type, e.g. 'acmt.001.001.08' "
"Account Opening Instruction. Must be exactly one of: "
f"{_MESSAGE_TYPE_LIST} (see list_message_types)."
),
json_schema_extra={"enum": _MESSAGE_TYPE_VALUES},
),
]
_IDENTIFIER_KIND_VALUES: list[str] = sorted(services._IDENTIFIER_VALIDATORS)
_IDENTIFIER_KIND_LIST = ", ".join(f"'{v}'" for v in _IDENTIFIER_KIND_VALUES)
_IdentifierKind = Annotated[
str,
Field(
description=(
"The financial identifier scheme to validate against "
"(case-insensitive). Must be exactly one of: "
f"{_IDENTIFIER_KIND_LIST}."
),
json_schema_extra={"enum": _IDENTIFIER_KIND_VALUES},
),
]
[docs]
def list_message_types() -> list[dict]:
"""List every supported ISO 20022 acmt 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.
``acmt.001.001.08`` Account Opening Instruction). Do not use it to fetch a
type's fields or schema -- call ``get_required_fields`` or
``get_input_schema`` for that.
Returns a list of ``{"message_type": ..., "name": ...}`` dictionaries, one
per supported message type (e.g. ``acmt.001.001.08``).
"""
try:
return services.list_message_types()
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 an acmt message type.
Use this for a quick checklist of the mandatory columns before building
account records. When you need full type/format constraints (not just
which fields are required), call ``get_input_schema`` instead.
Args:
message_type: A supported ISO 20022 acmt message type.
"""
try:
return services.get_required_fields(message_type)
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 account 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 account records against a message type's input 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 validate a single financial identifier in isolation use
``validate_identifier``.
Returns a report ``{"valid": bool, "total": int, "valid_count": int,
"errors": [...]}``.
Args:
message_type: A supported ISO 20022 acmt message type.
records: One or more flat account records to validate.
"""
try:
return services.validate_records(message_type, records)
except ValueError as exc:
return {"error": str(exc)}
[docs]
def validate_identifier(
kind: _IdentifierKind,
value: Annotated[
str,
Field(
description=(
"The identifier value to check, e.g. an IBAN, BIC/SWIFT code, "
"or LEI; validated according to the given kind."
)
),
],
) -> dict:
"""Validate a single financial identifier (IBAN, BIC, or LEI).
Use this for a one-off identifier check with a clear pass/fail. To
validate identifiers embedded across a whole batch of account records,
prefer ``validate_records`` rather than calling this per field.
Returns ``{"kind": str, "value": str, "valid": bool}``.
Args:
kind: One of ``"iban"``, ``"bic"``, or ``"lei"`` (case-insensitive).
value: The identifier value to check.
"""
try:
return services.validate_identifier(kind, value)
except ValueError as exc:
return {"error": str(exc)}
[docs]
def generate_message(
message_type: _MessageType,
records: Annotated[
list[dict],
Field(
description=(
"One or more flat account records, each a dict of field name "
"-> value, from which the acmt XML is generated; run "
"validate_records first to surface record-level errors."
)
),
],
) -> str:
"""Generate a validated ISO 20022 acmt XML message from in-memory records.
This is the primary generation tool: pass account 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 acmt message type.
records: One or more flat account records.
"""
try:
return services.generate(message_type, records)
except (ValueError, RuntimeError, OverflowError) as exc:
# The library raises RuntimeError when the rendered XML fails its
# XSD (OverflowError for a year its date types cannot hold); either is
# a validation error, which the SDK would otherwise hide behind a bare
# "Error executing tool". The schema's install path is noise.
return json.dumps({"error": _without_paths(str(exc))})
def _without_paths(message: str) -> str:
"""Replace absolute file paths in ``message`` with their basename."""
return re.sub(r"/\S*/([^/\s]+)", r"\1", message)
# A GLEIF record does not change from one minute to the next, but an agent
# often asks about the same LEI several times in one session, and each ask
# is a round trip to api.gleif.org (a second or two). Answers are kept for
# a short while; transport failures are not, so a retry reaches the API.
_LEI_CACHE_TTL_SECONDS = 300.0
_LEI_CACHE_MAX_ENTRIES = 256
_lei_cache: dict[str, tuple[float, dict]] = {}
def _lei_cache_get(lei_code: str) -> dict | None:
"""Return the cached answer for ``lei_code`` if it has not expired."""
entry = _lei_cache.get(lei_code)
if entry is None:
return None
expires_at, answer = entry
if time.monotonic() >= expires_at:
del _lei_cache[lei_code]
return None
return dict(answer)
def _lei_cache_put(lei_code: str, answer: dict) -> dict:
"""Remember ``answer`` for ``lei_code`` and return it."""
if len(_lei_cache) >= _LEI_CACHE_MAX_ENTRIES:
oldest = min(_lei_cache, key=lambda k: _lei_cache[k][0])
del _lei_cache[oldest]
_lei_cache[lei_code] = (
time.monotonic() + _LEI_CACHE_TTL_SECONDS,
dict(answer),
)
return dict(answer)
[docs]
def verify_lei_online(
lei_code: Annotated[
str,
Field(
description=(
"A 20-character ISO 17442 Legal Entity Identifier to look up "
"in the live GLEIF register, e.g. '5493001KJTIIGC8Y1R12'. Use "
"validate_identifier(kind='lei', ...) first for a purely "
"offline check-digit/format check."
)
),
],
) -> dict:
"""Verify a LEI against the live GLEIF register (reaches the network).
Unlike ``validate_identifier``, which checks an LEI's format and check
digits entirely offline, this tool performs a live HTTP GET against the
public GLEIF REST API (``api.gleif.org``) to confirm the LEI is actually
registered and to return the registered entity's details. It therefore
requires network access and is marked open-world.
Requires the optional ``online`` extra (``pip install acmt001-mcp[online]``)
for the ``httpx`` HTTP client; without it a graceful ``{"error": ...}`` is
returned rather than raising.
Returns a dict with the ``lei``, registered ``legal_name``, entity
``status``, ``country``, and registration ``registration_status``,
``initial_registration_date``, ``last_update_date`` and
``next_renewal_date``. On failure returns ``{"error": ...}``: an
``LEI not found`` error for an unregistered LEI (HTTP 404), or a
``GLEIF API unavailable`` error for any other non-200 response or a
transport/connection error.
Args:
lei_code: A 20-character ISO 17442 Legal Entity Identifier.
"""
try:
import httpx
except ImportError:
return {
"error": (
"verify_lei_online requires the optional 'online' extra; "
"install it with: pip install acmt001-mcp[online]"
)
}
cached = _lei_cache_get(lei_code)
if cached is not None:
return cached
url = _GLEIF_LEI_RECORD_URL.format(lei=lei_code)
try:
response = httpx.get(
url,
timeout=10.0,
headers={"Accept": "application/vnd.api+json"},
)
except httpx.HTTPError as exc:
return {"error": f"GLEIF API unavailable: {exc}"}
if response.status_code == 404:
return _lei_cache_put(
lei_code, {"error": f"LEI not found: {lei_code}"}
)
if response.status_code != 200:
return {"error": f"GLEIF API unavailable: HTTP {response.status_code}"}
attributes = response.json().get("data", {}).get("attributes", {})
entity = attributes.get("entity", {})
registration = attributes.get("registration", {})
return _lei_cache_put(
lei_code,
{
"lei": attributes.get("lei", lei_code),
"legal_name": entity.get("legalName", {}).get("name"),
"status": entity.get("status"),
"country": entity.get("legalAddress", {}).get("country"),
"registration_status": registration.get("status"),
"initial_registration_date": registration.get(
"initialRegistrationDate"
),
"last_update_date": registration.get("lastUpdateDate"),
"next_renewal_date": registration.get("nextRenewalDate"),
},
)
# ---------------------------------------------------------------------------
# Prompt.
#
# A guided workflow prompt that teaches an MCP client the canonical tool order
# for onboarding a corporate account. Like the tools, it is a pure function of
# its arguments -- it renders guidance text and never touches the backend.
# ---------------------------------------------------------------------------
[docs]
def onboard_corporate_account(
company_name: Annotated[
str,
Field(
description=(
"The legal name of the company being onboarded; left blank "
"for generic guidance."
)
),
] = "",
country: Annotated[
str,
Field(
description=(
"The ISO 3166-1 alpha-2 country code of the account owner, "
"e.g. 'GB'; left blank for generic guidance."
)
),
] = "",
) -> str:
"""Guide an agent through onboarding a corporate account end to end.
Renders a step-by-step playbook naming the tools to call and their order:
``list_message_types`` to pick a message type, then
``get_required_fields`` / ``get_input_schema`` to learn the input record,
then ``validate_records`` to check the batch, and finally
``generate_message`` to emit the validated acmt XML.
Args:
company_name: The legal name of the company being onboarded.
country: The ISO 3166-1 alpha-2 country code of the account owner.
Returns:
A guidance string describing the recommended tool workflow.
"""
subject = company_name.strip() or "a corporate account"
where = f" in {country.strip()}" if country.strip() else ""
return (
f"You are onboarding {subject}{where} as an ISO 20022 acmt account.\n"
"Follow this tool order:\n"
"1. Call list_message_types to choose the right acmt message type "
"(e.g. 'acmt.007.001.05' Account Opening Request).\n"
"2. Call get_required_fields for a quick checklist, then "
"get_input_schema for the full field types and constraints.\n"
"3. Assemble one flat record per account and call validate_records to "
"catch structural and identifier errors before generating.\n"
"4. Once validate_records reports valid, call generate_message to emit "
"the XSD-validated acmt XML document."
)
# ---------------------------------------------------------------------------
# Resources.
#
# Read-only views over the acmt001 catalogue, reusing the same ``services``
# facade functions as the tools. The static resource exposes the full message
# type catalogue; the templated resource describes a single message type's
# input record. Both return a ``json.dumps`` string, mirroring the tools'
# JSON-serializable contract.
# ---------------------------------------------------------------------------
[docs]
def message_types_resource() -> str:
"""Expose the full acmt message type catalogue as a JSON resource.
This is the resource form of ``list_message_types``: it returns the same
``[{"message_type": ..., "name": ...}, ...]`` catalogue, serialized as a
JSON string for clients that consume resources rather than tool calls.
Returns:
A JSON array string of ``{"message_type", "name"}`` objects.
"""
return json.dumps(services.list_message_types())
[docs]
def describe_message_type_resource(message_type: _MessageType) -> str:
"""Describe a single acmt message type's input record as a JSON resource.
This is the resource form of ``get_required_fields`` + ``get_input_schema``
for one message type: it returns both the required-field names and the full
input JSON Schema, serialized as a JSON string. On an unknown message type
it returns an ``{"error": ...}`` payload instead of raising, mirroring the
tools' error contract.
Args:
message_type: A supported ISO 20022 acmt message type.
Returns:
A JSON object string ``{"message_type", "required_fields",
"input_schema"}``, or ``{"error": ...}`` for an unknown type.
"""
try:
return json.dumps(
{
"message_type": message_type,
"required_fields": services.get_required_fields(message_type),
"input_schema": services.get_input_schema(message_type),
}
)
except ValueError as exc:
return json.dumps({"error": str(exc)})
# Tools, the prompt and the resources 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 the tools in this order.
server.tool(title="List acmt message types", annotations=_PURE_READ)(
list_message_types
)
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 IBAN, BIC or LEI", annotations=_PURE_READ)(
validate_identifier
)
server.tool(title="Generate acmt XML from records", annotations=_PURE_READ)(
generate_message
)
server.tool(
title="Verify a LEI against the live GLEIF register",
annotations=_EXTERNAL,
)(verify_lei_online)
server.prompt(title="Onboard a corporate account")(onboard_corporate_account)
server.resource(
"acmt001://message-types",
title="acmt message type catalogue",
mime_type="application/json",
)(message_types_resource)
server.resource(
"acmt001://describe/{message_type}",
title="Describe an acmt message type",
mime_type="application/json",
)(describe_message_type_resource)
[docs]
def main(argv: list[str] | None = None) -> None:
"""Run the acmt001 MCP server (the ``acmt001-mcp`` entry point).
stdio by default; ``--transport streamable-http`` or ``--transport sse``
listens on ``--host``/``--port`` instead. See :mod:`acmt001_mcp._cli`.
Args:
argv: Command-line arguments; ``None`` reads ``sys.argv[1:]``.
"""
_cli.serve(server, argv, "acmt001-mcp", __version__)
if __name__ == "__main__":
main()