API Reference

afas_mcp_server

afas_mcp_server: an MCP server that exposes AFAS Profit GetConnectors, UpdateConnectors and metainfo to AI agents.

Modules:

  • client –

    Thin asynchronous client for the AFAS Profit REST services.

  • configuration –

    Runtime configuration for the AFAS MCP server, read from AFAS_* environment variables.

  • filters –

    Filter, sort and paging expressions for GetConnector calls, rendered the way the AFAS REST API expects them.

  • server –

    The MCP server: tools that expose the connectors of one AFAS Profit environment to an AI agent.

  • validation –

    Offline checks of UpdateConnector payloads against the schema AFAS publishes for the connector.

Classes:

  • AfasApiError –

    An HTTP error response from the AFAS REST services, with the Profit error details decoded.

  • AfasClient –

    Asynchronous access to GetConnectors, UpdateConnectors and metainfo of one AFAS environment.

  • AfasError –

    Base class for errors raised by the AFAS client.

  • Environment –

    The kind of AFAS Online environment, which selects the REST hostname.

  • Filter –

    One condition on a GetConnector column.

  • Issue –

    One problem found in a payload.

  • Operator –

    Comparison operators AFAS supports in GetConnector filters, by readable name.

  • Settings –

    Connection and behaviour settings, populated from AFAS_* environment variables or a .env file.

Functions:

  • build_server –

    Create the MCP server for one AFAS environment.

  • validate_element –

    Check an Element payload against an UpdateConnector schema without calling AFAS.

AfasApiError 
AfasApiError(
    status_code: int,
    method: str,
    url: str,
    details: dict[str, Any],
)

Bases: AfasError


              flowchart TD
              afas_mcp_server.AfasApiError[AfasApiError]
              afas_mcp_server.client.AfasError[AfasError]

                              afas_mcp_server.client.AfasError --> afas_mcp_server.AfasApiError
                


              click afas_mcp_server.AfasApiError href "" "afas_mcp_server.AfasApiError"
              click afas_mcp_server.client.AfasError href "" "afas_mcp_server.client.AfasError"

An HTTP error response from the AFAS REST services, with the Profit error details decoded.

Parameters:

  • status_code
    (int) –

    The HTTP status code.

  • method
    (str) –

    The HTTP method of the failed call.

  • url
    (str) –

    The URL of the failed call.

  • details
    (dict[str, Any]) –

    Error information decoded from the body and the X-PROFIT-ERROR header.

Raised by: Used by:

Methods:

  • describe –

    Return a one-line human readable summary of the failure.

describe 
describe() -> str

Return a one-line human readable summary of the failure.

AfasClient 
AfasClient(
    settings: Settings,
    transport: AsyncBaseTransport | None = None,
)

Asynchronous access to GetConnectors, UpdateConnectors and metainfo of one AFAS environment.

Use it as an async context manager so the underlying HTTP connection pool is closed.

Parameters:

  • settings
    (Settings) –

    Connection settings.

  • transport
    (AsyncBaseTransport | None, default: None ) –

    Optional HTTP transport, used by tests to fake AFAS.

Used by:

Methods:

  • __aenter__ –

    Enter the context; the client is ready to use as soon as it is constructed.

  • __aexit__ –

    Close the HTTP connection pool.

  • aclose –

    Close the HTTP connection pool.

  • delete –

    Delete one record through an UpdateConnector.

  • describe_get_connector –

    Return the field definitions of a GetConnector.

  • describe_update_connector –

    Return the schema of an UpdateConnector, including nested objects.

  • fetch_rows –

    Return one page of rows from a GetConnector.

  • insert –

    Create records through an UpdateConnector.

  • list_connectors –

    Return the GetConnectors and UpdateConnectors the token may use, as AFAS reports them.

  • profit_version –

    Return the Profit version information, which doubles as a connectivity and token check.

  • request –

    Send one request to AFAS and return its decoded JSON body.

  • update –

    Change existing records through an UpdateConnector.

__aenter__ async 
__aenter__() -> Self

Enter the context; the client is ready to use as soon as it is constructed.

__aexit__ async 
__aexit__(*exc_info: object) -> None

Close the HTTP connection pool.

aclose async 
aclose() -> None

Close the HTTP connection pool.

delete async 
delete(
    connector_id: str,
    key_field: str,
    key_value: str | int,
    object_name: str | None = None,
) -> Any

Delete one record through an UpdateConnector.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

  • key_field (str) –

    The key field, with or without the leading @.

  • key_value (str | int) –

    The key value of the record to delete.

  • object_name (str | None, default: None ) –

    The element to delete from; defaults to the connector itself.

Returns:

  • Any –

    Whatever AFAS returns, often nothing.

describe_get_connector async 
describe_get_connector(connector_id: str) -> dict[str, Any]

Return the field definitions of a GetConnector.

Parameters:

  • connector_id (str) –

    The GetConnector id.

Returns:

  • dict[str, Any] –

    The metainfo document with name, description and fields.

describe_update_connector async 
describe_update_connector(
    connector_id: str,
) -> dict[str, Any]

Return the schema of an UpdateConnector, including nested objects.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

Returns:

  • dict[str, Any] –

    The metainfo document with fields and objects.

fetch_rows async 
fetch_rows(
    connector_id: str,
    *,
    skip: int = 0,
    take: int | None = None,
    filters: list[Filter] | None = None,
    order_by: list[str] | None = None,
) -> list[dict[str, Any]]

Return one page of rows from a GetConnector.

Parameters:

  • connector_id (str) –

    The GetConnector id.

  • skip (int, default: 0 ) –

    Number of rows to skip.

  • take (int | None, default: None ) –

    Number of rows to return; defaults to the configured page size.

  • filters (list[Filter] | None, default: None ) –

    Conditions to apply.

  • order_by (list[str] | None, default: None ) –

    Sort order; prefix a field with - for descending.

Returns:

  • list[dict[str, Any]] –

    The rows of the page, each a mapping of field id to value.

insert async 
insert(
    connector_id: str,
    element: dict[str, Any] | list[dict[str, Any]],
) -> Any

Create records through an UpdateConnector.

Parameters:

Returns:

  • Any –

    Whatever AFAS returns, usually the keys of the created records.

list_connectors async 
list_connectors() -> dict[str, Any]

Return the GetConnectors and UpdateConnectors the token may use, as AFAS reports them.

profit_version async 
profit_version() -> dict[str, Any]

Return the Profit version information, which doubles as a connectivity and token check.

request async 
request(
    method: str,
    path: str,
    *,
    params: dict[str, str] | None = None,
    body: dict[str, Any] | None = None,
) -> Any

Send one request to AFAS and return its decoded JSON body.

Parameters:

  • method (str) –

    HTTP method.

  • path (str) –

    Path relative to the REST services base URL.

  • params (dict[str, str] | None, default: None ) –

    Query parameters.

  • body (dict[str, Any] | None, default: None ) –

    JSON body for POST and PUT.

Returns:

  • Any –

    The decoded JSON response, or None when AFAS sent no body.

Raises:

  • AfasApiError –

    When AFAS answers with an HTTP error status.

update async 
update(
    connector_id: str,
    element: dict[str, Any] | list[dict[str, Any]],
) -> Any

Change existing records through an UpdateConnector.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

  • element (dict[str, Any] | list[dict[str, Any]]) –

    The Element payload, including the @Key entries that identify the records.

Returns:

  • Any –

    Whatever AFAS returns, often nothing.

AfasError

Bases: Exception


              flowchart TD
              afas_mcp_server.AfasError[AfasError]

              

              click afas_mcp_server.AfasError href "" "afas_mcp_server.AfasError"

Base class for errors raised by the AFAS client.

Subclassed by:
Environment

Bases: str, Enum


              flowchart TD
              afas_mcp_server.Environment[Environment]

              

              click afas_mcp_server.Environment href "" "afas_mcp_server.Environment"

The kind of AFAS Online environment, which selects the REST hostname.

Attributes:

  • host_prefix (str) –

    The hostname fragment AFAS Online uses for this environment kind.

host_prefix property 
host_prefix: str

The hostname fragment AFAS Online uses for this environment kind.

Filter

Bases: BaseModel


              flowchart TD
              afas_mcp_server.Filter[Filter]

              

              click afas_mcp_server.Filter href "" "afas_mcp_server.Filter"

One condition on a GetConnector column.

Returned by: Used by:

Attributes:

  • rendered_value (str) –

    The value as text, with the wildcards AFAS expects for the text operators.

rendered_value property 
rendered_value: str

The value as text, with the wildcards AFAS expects for the text operators.

Issue

Bases: BaseModel


              flowchart TD
              afas_mcp_server.Issue[Issue]

              

              click afas_mcp_server.Issue href "" "afas_mcp_server.Issue"

One problem found in a payload.

Returned by:
Operator

Bases: str, Enum


              flowchart TD
              afas_mcp_server.Operator[Operator]

              

              click afas_mcp_server.Operator href "" "afas_mcp_server.Operator"

Comparison operators AFAS supports in GetConnector filters, by readable name.

Attributes:

  • code (int) –

    The numeric operatortypes value AFAS uses for this operator.

code property 
code: int

The numeric operatortypes value AFAS uses for this operator.

Settings

Bases: BaseSettings


              flowchart TD
              afas_mcp_server.Settings[Settings]

              

              click afas_mcp_server.Settings href "" "afas_mcp_server.Settings"

Connection and behaviour settings, populated from AFAS_* environment variables or a .env file.

Either member_id (optionally with environment) or an explicit base_url must be provided.

Returned by: Used by:

Methods:

  • check_consistency –

    Reject settings that cannot produce a base URL or a usable log level.

Attributes:

authorization_header property 
authorization_header: str

The value of the Authorization header sent with every AFAS call.

python_log_level property 
python_log_level: str

The log level as Python's logging module spells it.

rest_base_url property 
rest_base_url: str

The REST services base URL, without a trailing slash.

check_consistency 
check_consistency() -> Self

Reject settings that cannot produce a base URL or a usable log level.

Returns:

  • Self –

    The validated settings, unchanged.

Raises:

  • ValueError –

    When neither an endpoint nor a known log level is available.

build_server 
build_server(
    settings: Settings | None = None,
    transport: AsyncBaseTransport | None = None,
) -> MCPServer

Create the MCP server for one AFAS environment.

Parameters:

  • settings
    (Settings | None, default: None ) –

    Connection settings; read from the environment when omitted.

  • transport
    (AsyncBaseTransport | None, default: None ) –

    Optional HTTP transport for the AFAS client, used by tests to fake AFAS.

Returns:

  • MCPServer –

    A server with the read tools, and the write tools when the settings allow writes.

validate_element 
validate_element(
    schema: dict[str, Any], element: object, operation: str
) -> list[Issue]

Check an Element payload against an UpdateConnector schema without calling AFAS.

Parameters:

  • schema
    (dict[str, Any]) –

    The UpdateConnector schema as returned by metainfo/update/{id}.

  • element
    (object) –

    The Element payload that would be sent; a mapping or a list of mappings.

  • operation
    (str) –

    insert or update.

Returns:

  • list[Issue] –

    The problems found; empty when the payload looks valid.

__main__

Command-line entry point, installed as afas-mcp-server and reachable as python -m afas_mcp_server.

Functions:

  • build_parser –

    Describe the command line.

  • configure_logging –

    Log to stderr; stdout belongs to the MCP protocol when running over stdio.

  • load_settings –

    Read the settings, explaining on stderr what is missing when they are invalid.

  • main –

    Run the command line.

  • profit_version –

    Ask AFAS for its version with a short-lived client.

  • run_check –

    Verify the connection and report the outcome.

  • serve –

    Build the server and run it on the requested transport until the client disconnects.

build_parser 
build_parser() -> ArgumentParser

Describe the command line.

configure_logging 
configure_logging(level: str) -> None

Log to stderr; stdout belongs to the MCP protocol when running over stdio.

load_settings 
load_settings() -> Settings | None

Read the settings, explaining on stderr what is missing when they are invalid.

Returns:

  • Settings | None –

    The settings, or None when they could not be validated.

main 
main(argv: Sequence[str] | None = None) -> int

Run the command line.

Parameters:

  • argv (Sequence[str] | None, default: None ) –

    Arguments without the program name; defaults to the process arguments.

Returns:

  • int –

    The process exit code.

profit_version async 
profit_version(settings: Settings) -> dict[str, Any]

Ask AFAS for its version with a short-lived client.

run_check 
run_check(settings: Settings) -> int

Verify the connection and report the outcome.

Parameters:

  • settings (Settings) –

    The connection settings to verify.

Returns:

  • int –

    The process exit code: 0 when AFAS answered, 1 otherwise.

serve 
serve(settings: Settings, args: Namespace) -> None

Build the server and run it on the requested transport until the client disconnects.

client

Thin asynchronous client for the AFAS Profit REST services.

Classes:

  • AfasApiError –

    An HTTP error response from the AFAS REST services, with the Profit error details decoded.

  • AfasClient –

    Asynchronous access to GetConnectors, UpdateConnectors and metainfo of one AFAS environment.

  • AfasError –

    Base class for errors raised by the AFAS client.

Functions:

  • connector_path –

    Join path segments under connectors/, percent-encoding everything but @.

  • decode_profit_error_header –

    Decode the base64 X-PROFIT-ERROR header AFAS adds to failed calls.

  • error_details –

    Collect everything AFAS tells us about a failed call.

  • parse_json –

    Return the JSON body of a response, or None when there is none.

  • update_body –

    Wrap an element in the envelope UpdateConnectors expect.

AfasApiError 
AfasApiError(
    status_code: int,
    method: str,
    url: str,
    details: dict[str, Any],
)

Bases: AfasError


              flowchart TD
              afas_mcp_server.client.AfasApiError[AfasApiError]
              afas_mcp_server.client.AfasError[AfasError]

                              afas_mcp_server.client.AfasError --> afas_mcp_server.client.AfasApiError
                


              click afas_mcp_server.client.AfasApiError href "" "afas_mcp_server.client.AfasApiError"
              click afas_mcp_server.client.AfasError href "" "afas_mcp_server.client.AfasError"

An HTTP error response from the AFAS REST services, with the Profit error details decoded.

Parameters:

  • status_code (int) –

    The HTTP status code.

  • method (str) –

    The HTTP method of the failed call.

  • url (str) –

    The URL of the failed call.

  • details (dict[str, Any]) –

    Error information decoded from the body and the X-PROFIT-ERROR header.

Raised by: Used by:

Methods:

  • describe –

    Return a one-line human readable summary of the failure.

describe 
describe() -> str

Return a one-line human readable summary of the failure.

AfasClient 
AfasClient(
    settings: Settings,
    transport: AsyncBaseTransport | None = None,
)

Asynchronous access to GetConnectors, UpdateConnectors and metainfo of one AFAS environment.

Use it as an async context manager so the underlying HTTP connection pool is closed.

Parameters:

  • settings (Settings) –

    Connection settings.

  • transport (AsyncBaseTransport | None, default: None ) –

    Optional HTTP transport, used by tests to fake AFAS.

Used by:

Methods:

  • __aenter__ –

    Enter the context; the client is ready to use as soon as it is constructed.

  • __aexit__ –

    Close the HTTP connection pool.

  • aclose –

    Close the HTTP connection pool.

  • delete –

    Delete one record through an UpdateConnector.

  • describe_get_connector –

    Return the field definitions of a GetConnector.

  • describe_update_connector –

    Return the schema of an UpdateConnector, including nested objects.

  • fetch_rows –

    Return one page of rows from a GetConnector.

  • insert –

    Create records through an UpdateConnector.

  • list_connectors –

    Return the GetConnectors and UpdateConnectors the token may use, as AFAS reports them.

  • profit_version –

    Return the Profit version information, which doubles as a connectivity and token check.

  • request –

    Send one request to AFAS and return its decoded JSON body.

  • update –

    Change existing records through an UpdateConnector.

__aenter__ async 
__aenter__() -> Self

Enter the context; the client is ready to use as soon as it is constructed.

__aexit__ async 
__aexit__(*exc_info: object) -> None

Close the HTTP connection pool.

aclose async 
aclose() -> None

Close the HTTP connection pool.

delete async 
delete(
    connector_id: str,
    key_field: str,
    key_value: str | int,
    object_name: str | None = None,
) -> Any

Delete one record through an UpdateConnector.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

  • key_field (str) –

    The key field, with or without the leading @.

  • key_value (str | int) –

    The key value of the record to delete.

  • object_name (str | None, default: None ) –

    The element to delete from; defaults to the connector itself.

Returns:

  • Any –

    Whatever AFAS returns, often nothing.

describe_get_connector async 
describe_get_connector(connector_id: str) -> dict[str, Any]

Return the field definitions of a GetConnector.

Parameters:

  • connector_id (str) –

    The GetConnector id.

Returns:

  • dict[str, Any] –

    The metainfo document with name, description and fields.

describe_update_connector async 
describe_update_connector(
    connector_id: str,
) -> dict[str, Any]

Return the schema of an UpdateConnector, including nested objects.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

Returns:

  • dict[str, Any] –

    The metainfo document with fields and objects.

fetch_rows async 
fetch_rows(
    connector_id: str,
    *,
    skip: int = 0,
    take: int | None = None,
    filters: list[Filter] | None = None,
    order_by: list[str] | None = None,
) -> list[dict[str, Any]]

Return one page of rows from a GetConnector.

Parameters:

  • connector_id (str) –

    The GetConnector id.

  • skip (int, default: 0 ) –

    Number of rows to skip.

  • take (int | None, default: None ) –

    Number of rows to return; defaults to the configured page size.

  • filters (list[Filter] | None, default: None ) –

    Conditions to apply.

  • order_by (list[str] | None, default: None ) –

    Sort order; prefix a field with - for descending.

Returns:

  • list[dict[str, Any]] –

    The rows of the page, each a mapping of field id to value.

insert async 
insert(
    connector_id: str,
    element: dict[str, Any] | list[dict[str, Any]],
) -> Any

Create records through an UpdateConnector.

Parameters:

Returns:

  • Any –

    Whatever AFAS returns, usually the keys of the created records.

list_connectors async 
list_connectors() -> dict[str, Any]

Return the GetConnectors and UpdateConnectors the token may use, as AFAS reports them.

profit_version async 
profit_version() -> dict[str, Any]

Return the Profit version information, which doubles as a connectivity and token check.

request async 
request(
    method: str,
    path: str,
    *,
    params: dict[str, str] | None = None,
    body: dict[str, Any] | None = None,
) -> Any

Send one request to AFAS and return its decoded JSON body.

Parameters:

  • method (str) –

    HTTP method.

  • path (str) –

    Path relative to the REST services base URL.

  • params (dict[str, str] | None, default: None ) –

    Query parameters.

  • body (dict[str, Any] | None, default: None ) –

    JSON body for POST and PUT.

Returns:

  • Any –

    The decoded JSON response, or None when AFAS sent no body.

Raises:

  • AfasApiError –

    When AFAS answers with an HTTP error status.

update async 
update(
    connector_id: str,
    element: dict[str, Any] | list[dict[str, Any]],
) -> Any

Change existing records through an UpdateConnector.

Parameters:

  • connector_id (str) –

    The UpdateConnector id.

  • element (dict[str, Any] | list[dict[str, Any]]) –

    The Element payload, including the @Key entries that identify the records.

Returns:

  • Any –

    Whatever AFAS returns, often nothing.

AfasError

Bases: Exception


              flowchart TD
              afas_mcp_server.client.AfasError[AfasError]

              

              click afas_mcp_server.client.AfasError href "" "afas_mcp_server.client.AfasError"

Base class for errors raised by the AFAS client.

Subclassed by:
connector_path 
connector_path(*segments: str) -> str

Join path segments under connectors/, percent-encoding everything but @.

Parameters:

  • segments (str, default: () ) –

    The path segments after connectors.

Returns:

  • str –

    The relative request path.

decode_profit_error_header 
decode_profit_error_header(value: str) -> dict[str, Any]

Decode the base64 X-PROFIT-ERROR header AFAS adds to failed calls.

Parameters:

  • value (str) –

    The raw header value.

Returns:

  • dict[str, Any] –

    The decoded JSON object, or the decoded text under profitError when it is not JSON.

error_details 
error_details(response: Response) -> dict[str, Any]

Collect everything AFAS tells us about a failed call.

Parameters:

  • response (Response) –

    The error response.

Returns:

  • dict[str, Any] –

    The X-PROFIT-ERROR header and JSON body merged, or the plain text body under message.

parse_json 
parse_json(response: Response) -> Any

Return the JSON body of a response, or None when there is none.

Parameters:

  • response (Response) –

    The HTTP response.

Returns:

  • Any –

    The decoded JSON value, or None for an empty or non-JSON body.

update_body 
update_body(
    connector_id: str,
    element: dict[str, Any] | list[dict[str, Any]],
) -> dict[str, Any]

Wrap an element in the envelope UpdateConnectors expect.

Parameters:

  • connector_id (str) –

    The UpdateConnector id, e.g. KnSubject.

  • element (dict[str, Any] | list[dict[str, Any]]) –

    The Element payload with Fields and optional Objects.

Returns:

  • dict[str, Any] –

    The request body {connector_id: {"Element": element}}.

configuration

Runtime configuration for the AFAS MCP server, read from AFAS_* environment variables.

Classes:

  • Environment –

    The kind of AFAS Online environment, which selects the REST hostname.

  • Settings –

    Connection and behaviour settings, populated from AFAS_* environment variables or a .env file.

Functions:

  • encode_token –

    Return the base64 token AFAS expects after AfasToken in the Authorization header.

  • is_encoded_token –

    Return whether candidate is already the base64 form of a <token> XML document.

Environment

Bases: str, Enum


              flowchart TD
              afas_mcp_server.configuration.Environment[Environment]

              

              click afas_mcp_server.configuration.Environment href "" "afas_mcp_server.configuration.Environment"

The kind of AFAS Online environment, which selects the REST hostname.

Attributes:

  • host_prefix (str) –

    The hostname fragment AFAS Online uses for this environment kind.

host_prefix property 
host_prefix: str

The hostname fragment AFAS Online uses for this environment kind.

Settings

Bases: BaseSettings


              flowchart TD
              afas_mcp_server.configuration.Settings[Settings]

              

              click afas_mcp_server.configuration.Settings href "" "afas_mcp_server.configuration.Settings"

Connection and behaviour settings, populated from AFAS_* environment variables or a .env file.

Either member_id (optionally with environment) or an explicit base_url must be provided.

Returned by: Used by:

Methods:

  • check_consistency –

    Reject settings that cannot produce a base URL or a usable log level.

Attributes:

authorization_header property 
authorization_header: str

The value of the Authorization header sent with every AFAS call.

python_log_level property 
python_log_level: str

The log level as Python's logging module spells it.

rest_base_url property 
rest_base_url: str

The REST services base URL, without a trailing slash.

check_consistency 
check_consistency() -> Self

Reject settings that cannot produce a base URL or a usable log level.

Returns:

  • Self –

    The validated settings, unchanged.

Raises:

  • ValueError –

    When neither an endpoint nor a known log level is available.

encode_token 
encode_token(token: str) -> str

Return the base64 token AFAS expects after AfasToken in the Authorization header.

Accepts the token in any of the forms AFAS hands out: the full <token> XML, only the value of its <data> element, or the XML already base64 encoded.

Parameters:

  • token (str) –

    The app connector token in one of the accepted forms.

Returns:

  • str –

    The base64 encoded <token> XML.

is_encoded_token 
is_encoded_token(candidate: str) -> bool

Return whether candidate is already the base64 form of a <token> XML document.

Parameters:

  • candidate (str) –

    The text to inspect.

Returns:

  • bool –

    True when the text base64-decodes to something that starts with <token.

filters

Filter, sort and paging expressions for GetConnector calls, rendered the way the AFAS REST API expects them.

Classes:

  • Filter –

    One condition on a GetConnector column.

  • Operator –

    Comparison operators AFAS supports in GetConnector filters, by readable name.

Functions:

  • compact_filter_params –

    Render filters as the filterfieldids/filtervalues/operatortypes query parameters.

  • filter_groups –

    Bucket filters by group number, in ascending group order.

  • filter_params –

    Render filters as query parameters, picking the compact syntax unless a separator forces JSON.

  • json_filter_params –

    Render filters as the filterjson query parameter.

  • needs_json_filter –

    Return whether any field or value contains a separator, which the compact syntax cannot escape.

  • order_by_param –

    Render the sort order as the orderbyfieldids query parameter.

  • query_params –

    Build the complete query string for a GetConnector call.

Filter

Bases: BaseModel


              flowchart TD
              afas_mcp_server.filters.Filter[Filter]

              

              click afas_mcp_server.filters.Filter href "" "afas_mcp_server.filters.Filter"

One condition on a GetConnector column.

Returned by: Used by:

Attributes:

  • rendered_value (str) –

    The value as text, with the wildcards AFAS expects for the text operators.

rendered_value property 
rendered_value: str

The value as text, with the wildcards AFAS expects for the text operators.

Operator

Bases: str, Enum


              flowchart TD
              afas_mcp_server.filters.Operator[Operator]

              

              click afas_mcp_server.filters.Operator href "" "afas_mcp_server.filters.Operator"

Comparison operators AFAS supports in GetConnector filters, by readable name.

Attributes:

  • code (int) –

    The numeric operatortypes value AFAS uses for this operator.

code property 
code: int

The numeric operatortypes value AFAS uses for this operator.

compact_filter_params 
compact_filter_params(
    groups: list[list[Filter]],
) -> dict[str, str]

Render filters as the filterfieldids/filtervalues/operatortypes query parameters.

Parameters:

Returns:

  • dict[str, str] –

    The three query parameters, with , between AND conditions and ; between OR groups.

filter_groups 
filter_groups(filters: list[Filter]) -> list[list[Filter]]

Bucket filters by group number, in ascending group order.

Parameters:

Returns:

  • list[list[Filter]] –

    One list per group; filters within a list are ANDed, lists are ORed.

filter_params 
filter_params(
    filters: list[Filter] | None,
) -> dict[str, str]

Render filters as query parameters, picking the compact syntax unless a separator forces JSON.

Parameters:

  • filters (list[Filter] | None) –

    The conditions to render, or None for no filtering.

Returns:

  • dict[str, str] –

    Query parameters to add to the GetConnector request.

json_filter_params 
json_filter_params(
    groups: list[list[Filter]],
) -> dict[str, str]

Render filters as the filterjson query parameter.

Parameters:

Returns:

  • dict[str, str] –

    A single filterjson parameter with one Filter entry per OR group.

needs_json_filter 
needs_json_filter(filters: list[Filter]) -> bool

Return whether any field or value contains a separator, which the compact syntax cannot escape.

Parameters:

Returns:

  • bool –

    True when the JSON filter syntax must be used.

order_by_param 
order_by_param(
    order_by: list[str] | None,
) -> dict[str, str]

Render the sort order as the orderbyfieldids query parameter.

Parameters:

  • order_by (list[str] | None) –

    Field ids to sort on; prefix with - for descending.

Returns:

  • dict[str, str] –

    The query parameter, or nothing when no sort order was given.

query_params 
query_params(
    *,
    skip: int,
    take: int,
    filters: list[Filter] | None = None,
    order_by: list[str] | None = None,
) -> dict[str, str]

Build the complete query string for a GetConnector call.

Parameters:

  • skip (int) –

    Number of rows to skip.

  • take (int) –

    Number of rows to return.

  • filters (list[Filter] | None, default: None ) –

    Conditions to apply, if any.

  • order_by (list[str] | None, default: None ) –

    Sort order, if any.

Returns:

  • dict[str, str] –

    All query parameters for the request.

server

The MCP server: tools that expose the connectors of one AFAS Profit environment to an AI agent.

Classes:

  • AppContext –

    What every tool needs at request time.

Functions:

  • app_context –

    Return the shared settings and client that the lifespan created for this server.

  • build_server –

    Create the MCP server for one AFAS environment.

  • call_afas –

    Await an AFAS call, translating its errors into tool errors the model can read.

  • check_page –

    Reject paging parameters AFAS or the configured limits would not accept.

  • connection_info –

    Show which AFAS environment this server talks to and confirm the token works by asking for the Profit version.

  • delete –

    Permanently delete one record in AFAS through an UpdateConnector (HTTP DELETE).

  • describe_failure –

    Turn an AFAS error into a message the model can act on, including the internal detail when present.

  • describe_get_connector –

    Describe a GetConnector: its field ids, labels, data types and lengths. Use the ids in filters and order_by.

  • describe_update_connector –

    Describe an UpdateConnector: its fields (mandatory, key, allowed values) and the nested objects it accepts.

  • ensure_valid –

    Fetch the connector schema and refuse the payload when it has problems.

  • get_rows –

    Read one page of rows from a GetConnector, with optional filters and sort order.

  • insert –

    Create records in AFAS through an UpdateConnector (HTTP POST).

  • instructions_for –

    Compose the guidance the MCP client shows its model, depending on whether writes are enabled.

  • list_connectors –

    List the GetConnectors and UpdateConnectors the configured token may use, optionally filtered by keyword.

  • matching –

    Keep the connectors whose id or description contains the search text, case-insensitively.

  • package_version –

    Return the installed version of this package, or 0.0.0 when it is not installed.

  • register_read_tools –

    Register the tools that never change data in AFAS.

  • register_write_tools –

    Register the tools that create, change or delete data in AFAS.

  • update –

    Change existing records in AFAS through an UpdateConnector (HTTP PUT); identify them with "@KeyField" entries.

  • validate_payload –

    Check an UpdateConnector payload against the connector schema without sending it, and show the exact body.

AppContext dataclass 
AppContext(settings: Settings, client: AfasClient)

What every tool needs at request time.

Returned by:
app_context 
app_context(ctx: ToolContext) -> AppContext

Return the shared settings and client that the lifespan created for this server.

build_server 
build_server(
    settings: Settings | None = None,
    transport: AsyncBaseTransport | None = None,
) -> MCPServer

Create the MCP server for one AFAS environment.

Parameters:

  • settings (Settings | None, default: None ) –

    Connection settings; read from the environment when omitted.

  • transport (AsyncBaseTransport | None, default: None ) –

    Optional HTTP transport for the AFAS client, used by tests to fake AFAS.

Returns:

  • MCPServer –

    A server with the read tools, and the write tools when the settings allow writes.

call_afas async 
call_afas(operation: Awaitable[T]) -> T

Await an AFAS call, translating its errors into tool errors the model can read.

Parameters:

  • operation (Awaitable[T]) –

    The pending client call.

Returns:

  • T –

    The result of the call.

Raises:

  • ToolError –

    When AFAS rejected the call.

check_page 
check_page(
    skip: int, take: int, settings: Settings
) -> None

Reject paging parameters AFAS or the configured limits would not accept.

Raises:

  • ToolError –

    When skip is negative or take is outside 1..max_take.

connection_info async 
connection_info(ctx: ToolContext) -> dict[str, Any]

Show which AFAS environment this server talks to and confirm the token works by asking for the Profit version.

delete async 
delete(
    ctx: ToolContext,
    connector_id: UpdateConnectorId,
    key_field: Annotated[
        str,
        Field(
            description='Primary key field id of the record, e.g. SbId.'
        ),
    ],
    key_value: Annotated[
        str | int,
        Field(
            description='Value of the key field for the record to delete.'
        ),
    ],
    *,
    object_name: Annotated[
        str | None,
        Field(
            description='Nested object to delete from; defaults to the connector itself.'
        ),
    ] = None,
) -> dict[str, Any]

Permanently delete one record in AFAS through an UpdateConnector (HTTP DELETE).

describe_failure 
describe_failure(error: AfasApiError) -> str

Turn an AFAS error into a message the model can act on, including the internal detail when present.

describe_get_connector async 
describe_get_connector(
    ctx: ToolContext, connector_id: ConnectorId
) -> dict[str, Any]

Describe a GetConnector: its field ids, labels, data types and lengths. Use the ids in filters and order_by.

describe_update_connector async 
describe_update_connector(
    ctx: ToolContext, connector_id: UpdateConnectorId
) -> dict[str, Any]

Describe an UpdateConnector: its fields (mandatory, key, allowed values) and the nested objects it accepts.

ensure_valid async 
ensure_valid(
    client: AfasClient,
    connector_id: str,
    element: Element,
    operation: Operation,
) -> None

Fetch the connector schema and refuse the payload when it has problems.

Raises:

  • ToolError –

    Listing every problem found, so the model can fix them all at once.

get_rows async 
get_rows(
    ctx: ToolContext,
    connector_id: ConnectorId,
    *,
    skip: Annotated[
        int,
        Field(
            description='Number of rows to skip; combine with take to page.'
        ),
    ] = 0,
    take: Annotated[
        int | None,
        Field(
            description='Rows to return; defaults to the configured page size.'
        ),
    ] = None,
    filters: Annotated[
        list[Filter] | None,
        Field(
            description='Conditions on field ids. Same group = AND, different groups = OR.'
        ),
    ] = None,
    order_by: Annotated[
        list[str] | None,
        Field(
            description='Field ids to sort on, prefix with - for descending. Required for reliable paging.'
        ),
    ] = None,
) -> dict[str, Any]

Read one page of rows from a GetConnector, with optional filters and sort order.

insert async 
insert(
    ctx: ToolContext,
    connector_id: UpdateConnectorId,
    element: ElementPayload,
    *,
    validate: Annotated[
        bool,
        Field(
            description='Check the payload against the schema before sending.'
        ),
    ] = True,
) -> dict[str, Any]

Create records in AFAS through an UpdateConnector (HTTP POST).

instructions_for 
instructions_for(settings: Settings) -> str

Compose the guidance the MCP client shows its model, depending on whether writes are enabled.

list_connectors async 
list_connectors(
    ctx: ToolContext,
    *,
    kind: Annotated[
        ConnectorKind,
        Field(
            description='Which connectors to list: get (read), update (write) or all.'
        ),
    ] = 'all',
    search: Annotated[
        str | None,
        Field(
            description='Keep only connectors whose id or description contains this.'
        ),
    ] = None,
) -> dict[str, Any]

List the GetConnectors and UpdateConnectors the configured token may use, optionally filtered by keyword.

matching 
matching(
    items: list[dict[str, Any]], search: str | None
) -> list[dict[str, Any]]

Keep the connectors whose id or description contains the search text, case-insensitively.

package_version 
package_version() -> str

Return the installed version of this package, or 0.0.0 when it is not installed.

register_read_tools 
register_read_tools(server: MCPServer) -> None

Register the tools that never change data in AFAS.

register_write_tools 
register_write_tools(server: MCPServer) -> None

Register the tools that create, change or delete data in AFAS.

update async 
update(
    ctx: ToolContext,
    connector_id: UpdateConnectorId,
    element: ElementPayload,
    *,
    validate: Annotated[
        bool,
        Field(
            description='Check the payload against the schema before sending.'
        ),
    ] = True,
) -> dict[str, Any]

Change existing records in AFAS through an UpdateConnector (HTTP PUT); identify them with "@KeyField" entries.

validate_payload async 
validate_payload(
    ctx: ToolContext,
    connector_id: UpdateConnectorId,
    element: ElementPayload,
    *,
    operation: Annotated[
        Operation,
        Field(
            description='insert requires all mandatory fields; update does not.'
        ),
    ] = 'insert',
) -> dict[str, Any]

Check an UpdateConnector payload against the connector schema without sending it, and show the exact body.

validation

Offline checks of UpdateConnector payloads against the schema AFAS publishes for the connector.

Classes:

  • Issue –

    One problem found in a payload.

Functions:

  • element_issues –

    Yield every problem in an Element payload, recursing into nested objects.

  • field_value_issues –

    Yield problems with one field value: type, length and allowed values.

  • fields_by_id –

    Index the fields of a schema node by field id.

  • fields_issues –

    Yield problems with the Fields mapping of a record.

  • is_integer –

    Return whether value is an int and not a bool.

  • is_number –

    Return whether value is an int or float and not a bool.

  • join_path –

    Join non-empty path parts with dots.

  • key_issues –

    Yield problems with the @Key entries that identify a record.

  • mandatory_issues –

    Yield the mandatory fields an insert leaves out.

  • object_issues –

    Yield problems inside one nested object: unknown name, missing Element, or issues in that element.

  • objects_by_name –

    Index the nested objects of a schema node by name.

  • objects_issues –

    Yield problems inside Objects, which AFAS accepts as a mapping or as a list of single-key mappings.

  • unknown_key_issues –

    Yield element keys that are neither @Key entries nor Fields/Objects.

  • validate_element –

    Check an Element payload against an UpdateConnector schema without calling AFAS.

Issue

Bases: BaseModel


              flowchart TD
              afas_mcp_server.validation.Issue[Issue]

              

              click afas_mcp_server.validation.Issue href "" "afas_mcp_server.validation.Issue"

One problem found in a payload.

Returned by:
element_issues 
element_issues(
    schema: dict[str, Any],
    element: object,
    operation: str,
    path: str = '',
) -> Iterator[Issue]

Yield every problem in an Element payload, recursing into nested objects.

Parameters:

  • schema (dict[str, Any]) –

    The connector or nested object schema from metainfo.

  • element (object) –

    The Element value; anything but a mapping or a list of mappings is itself reported.

  • operation (str) –

    insert or update; only inserts must supply all mandatory fields.

  • path (str, default: '' ) –

    Where the element sits in the payload.

field_value_issues 
field_value_issues(
    field: dict[str, Any], value: object, path: str
) -> Iterator[Issue]

Yield problems with one field value: type, length and allowed values.

Parameters:

  • field (dict[str, Any]) –

    The field definition from the schema.

  • value (object) –

    The supplied value.

  • path (str) –

    Where the value sits in the payload.

fields_by_id 
fields_by_id(
    schema: dict[str, Any],
) -> dict[str, dict[str, Any]]

Index the fields of a schema node by field id.

fields_issues 
fields_issues(
    element: dict[str, Any],
    fields: dict[str, dict[str, Any]],
    path: str,
) -> Iterator[Issue]

Yield problems with the Fields mapping of a record.

is_integer 
is_integer(value: object) -> bool

Return whether value is an int and not a bool.

is_number 
is_number(value: object) -> bool

Return whether value is an int or float and not a bool.

join_path 
join_path(*parts: str) -> str

Join non-empty path parts with dots.

key_issues 
key_issues(
    element: dict[str, Any],
    fields: dict[str, dict[str, Any]],
    path: str,
) -> Iterator[Issue]

Yield problems with the @Key entries that identify a record.

mandatory_issues 
mandatory_issues(
    element: dict[str, Any],
    fields: dict[str, dict[str, Any]],
    path: str,
) -> Iterator[Issue]

Yield the mandatory fields an insert leaves out.

object_issues 
object_issues(
    name: str,
    node: object,
    objects: dict[str, dict[str, Any]],
    operation: str,
    object_path: str,
) -> Iterator[Issue]

Yield problems inside one nested object: unknown name, missing Element, or issues in that element.

objects_by_name 
objects_by_name(
    schema: dict[str, Any],
) -> dict[str, dict[str, Any]]

Index the nested objects of a schema node by name.

objects_issues 
objects_issues(
    element: dict[str, Any],
    objects: dict[str, dict[str, Any]],
    operation: str,
    path: str,
) -> Iterator[Issue]

Yield problems inside Objects, which AFAS accepts as a mapping or as a list of single-key mappings.

unknown_key_issues 
unknown_key_issues(
    element: dict[str, Any], path: str
) -> Iterator[Issue]

Yield element keys that are neither @Key entries nor Fields/Objects.

validate_element 
validate_element(
    schema: dict[str, Any], element: object, operation: str
) -> list[Issue]

Check an Element payload against an UpdateConnector schema without calling AFAS.

Parameters:

  • schema (dict[str, Any]) –

    The UpdateConnector schema as returned by metainfo/update/{id}.

  • element (object) –

    The Element payload that would be sent; a mapping or a list of mappings.

  • operation (str) –

    insert or update.

Returns:

  • list[Issue] –

    The problems found; empty when the payload looks valid.