API Reference

This page provides detailed API documentation for the Rossum MCP Server.

Server Module

rossum_mcp.server.create_app()

Create and configure the MCP server.

Reads configuration from environment variables:

• ROSSUM_API_BASE_URL (required)

• ROSSUM_API_TOKEN (required)

• ROSSUM_MCP_MODE (optional, default: read-write)

• ROSSUM_MCP_LOG_LEVEL (optional, default: INFO)

Return type:

FastMCP

rossum_mcp.server.main()

Main entry point for console script.

Return type:

None

Base Utilities

class rossum_mcp.tools.base.GracefulListResult(items, skipped_ids=<factory>)

Bases: Generic[T]

items: list[T]

skipped_ids: list[int | str]

__init__(items, skipped_ids=<factory>)

rossum_mcp.tools.base.extract_id_from_url(url)

Extract the integer resource ID from a Rossum API URL.

Return type:

int

rossum_mcp.tools.base.build_resource_url(base_url, resource_type, resource_id)

Build a full URL for a Rossum API resource.

Return type:

str

rossum_mcp.tools.base.build_filters(**kwargs)

Build a filter dict from kwargs, excluding None values.

Return type:

dict[str, Any]

rossum_mcp.tools.base.filter_by_name_regex(items, name, use_regex)

Apply client-side regex name filtering when use_regex is True.

Return type:

list[TypeVar(T)]

async rossum_mcp.tools.base.graceful_list(client, resource, resource_label, max_items=None, **filters)

List resources gracefully, skipping items that fail deserialization.

Uses _http_client.fetch_all directly so that a single broken item does not terminate the entire iteration (the high-level client generators die on the first deserialization error).

Return type:

GracefulListResult

async rossum_mcp.tools.base.delete_resource(resource_type, resource_id, delete_fn, success_message=None)

Generic delete operation.

Write-access is enforced at the MCP layer via tags={“write”} + mcp.disable().

Parameters:

• resource_type (str) – Name of the resource (e.g., “queue”, “workspace”)

• resource_id (int) – ID of the resource to delete

• delete_fn (Callable[[int], Awaitable[None]]) – Async function that performs the deletion

• success_message (str | None) – Custom success message. If None, uses default format.

Return type:

dict

Tool Catalog & Discovery

Tool Catalog

Category metadata for dynamic tool discovery.

Provides descriptions and keywords for tool categories. Tool membership and read_only status are derived from tags set on individual @mcp.tool decorators.

class rossum_mcp.tools.catalog.CategoryMeta(description, keywords)

Bases: object

Lightweight metadata for a tool category.

description: str

keywords: list[str]

__init__(description, keywords)

rossum_mcp.tools.catalog.get_catalog_summary()

Get a compact text summary of all tool categories for the system prompt.

Return type:

str

Discovery

Discovery tools for dynamic tool loading.

Provides MCP tool to explore available tool categories and their metadata. Tool lists are derived from tags on @mcp.tool decorators rather than a static catalog.

rossum_mcp.tools.discovery.register_discovery_tools(mcp)

Return type:

None

Resource Tracking

rossum_mcp.tools.resource_tracking.track_resource(tracked, entity_type, entity_id, data)

Append a side-effect resource to the tracked list.

Converts dataclass/dict data to a plain dict for serialization. Non-convertible data is silently skipped.

Return type:

None

rossum_mcp.tools.resource_tracking.embed_tracked_resources(result, tracked)

Embed tracked resources into the tool result if any were collected.

If tracked is non-empty, converts the result to a dict (via dataclass asdict if needed) and adds the _tracked_resources key. Returns result unchanged if tracked is empty or result is not convertible.

Return type:

Any

Create Operations

rossum_mcp.tools.create.handler.register_create_tools(mcp, client, base_url)

Return type:

None

Get Operations

class rossum_mcp.tools.get.handler.EntityType(*values)

Bases: StrEnum

QUEUE = 'queue'

SCHEMA = 'schema'

HOOK = 'hook'

ENGINE = 'engine'

RULE = 'rule'

USER = 'user'

WORKSPACE = 'workspace'

EMAIL_TEMPLATE = 'email_template'

ORGANIZATION_GROUP = 'organization_group'

ANNOTATION = 'annotation'

RELATION = 'relation'

DOCUMENT_RELATION = 'document_relation'

ORGANIZATION_LIMIT = 'organization_limit'

HOOK_SECRETS_KEYS = 'hook_secrets_keys'

rossum_mcp.tools.get.handler.register_get_tools(mcp, client)

Return type:

None

class rossum_mcp.tools.get.registry.EntityConfig(retrieve_fn, search_fn)

Bases: object

retrieve_fn: Callable[[...], Awaitable[object]] | None

search_fn: Callable[[...], Awaitable[list]] | None

__init__(retrieve_fn, search_fn)

rossum_mcp.tools.get.registry.build_get_registry(client)

Build registry with retrieve functions only (search_fn populated by search layer).

Return type:

dict[str, EntityConfig]

class rossum_mcp.tools.get.models.SchemaTreeNode(id, label, category, type=None, required=False, hidden=False, children=None)

Bases: object

Lightweight schema node for tree structure display.

id: str

label: str

category: str

type: str | None = None

required: bool = False

hidden: bool = False

children: list[SchemaTreeNode] | None = None

to_dict()

Return type:

dict

__init__(id, label, category, type=None, required=False, hidden=False, children=None)

rossum_mcp.tools.get.schemas.extract_schema_tree(schema)

Return type:

list[dict]

include_related fetchers for enriched get responses.

Only meaningful for a few entities. Uses asyncio.gather for parallel fetches.

async rossum_mcp.tools.get.related.fetch_related(client, entity, entity_id, obj=None)

Fetch related data for an entity. Returns None if no fetcher exists.

Return type:

dict[str, object] | None

Search Operations

type rossum_mcp.tools.search.registry.Timestamp = Annotated[str, "ISO 8601 timestamp (e.g., '2024-01-15T10:30:00Z')"]

rossum_mcp.tools.search.registry.build_search_registry(client)

Build a flat dict of entity -> search function.

Return type:

dict[str, Callable[..., Awaitable[list]] | None]

rossum_mcp.tools.search.registry.extract_search_kwargs(query)

Extract filter kwargs from a search query model, dropping the entity discriminator.

Return type:

dict[str, object]

Pydantic search query models for the unified search layer.

Each entity type has specific filter fields. The discriminated union on entity produces a JSON Schema oneOf, so the LLM sees valid filters per entity type.

class rossum_mcp.tools.search.models.QueueSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['queue']

id: str | None

workspace_id: int | None

name: str | None

use_regex: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.SchemaSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['schema']

name: str | None

queue_id: int | None

use_regex: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.HookSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['hook']

queue_id: int | None

active: bool | None

first_n: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.EngineSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['engine']

id: int | None

engine_type: EngineType | None

agenda_id: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.RuleSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['rule']

schema_id: int | None

organization_id: int | None

enabled: bool | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.UserSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['user']

username: str | None

email: str | None

first_name: str | None

last_name: str | None

is_active: bool | None

is_organization_group_admin: bool | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.WorkspaceSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['workspace']

organization_id: int | None

name: str | None

use_regex: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.EmailTemplateSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['email_template']

queue_id: int | None

type: EmailTemplateType | None

name: str | None

first_n: int | None

use_regex: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.OrganizationGroupSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['organization_group']

name: str | None

use_regex: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.AnnotationSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['annotation']

queue_id: int

status: str | None

ordering: Sequence[str] | None

first_n: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.RelationSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['relation']

id: int | None

type: str | None

parent: int | None

key: str | None

annotation: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.DocumentRelationSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['document_relation']

id: int | None

type: str | None

annotation: int | None

key: str | None

documents: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.HookLogSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['hook_log']

hook_id: int | None

queue_id: int | None

annotation_id: int | None

email_id: int | None

log_level: list[LogLevel] | LogLevel | None

status: str | None

status_code: int | None

request_id: str | None

timestamp_before: str | None

timestamp_after: str | None

start_before: str | None

start_after: str | None

end_before: str | None

end_after: str | None

search: str | None

page_size: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.HookTemplateSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['hook_template']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.UserRoleSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['user_role']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.QueueTemplateNameSearch(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

entity: Literal['queue_template_name']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class rossum_mcp.tools.search.models.QueueListItem(id, name, url, workspace=None, schema=None, inbox=None, connector=None, automation_enabled=False, automation_level='never', status=None, counts=None, settings='<omitted>')

Bases: object

Queue summary for list responses (settings omitted to save context).

id: int

name: str

url: str

workspace: str | None = None

schema: str | None = None

inbox: str | None = None

connector: str | None = None

automation_enabled: bool = False

automation_level: str = 'never'

status: str | None = None

counts: dict[str, int] | None = None

settings: str = '<omitted>'

__init__(id, name, url, workspace=None, schema=None, inbox=None, connector=None, automation_enabled=False, automation_level='never', status=None, counts=None, settings='<omitted>')

class rossum_mcp.tools.search.models.SchemaListItem(id, name=None, queues=None, url=None, content='<omitted>', metadata=None, modified_by=None, modified_at=None)

Bases: object

Schema summary for list responses (content omitted to save context).

id: int

name: str | None = None

queues: list[str] | None = None

url: str | None = None

content: str = '<omitted>'

metadata: dict | None = None

modified_by: str | None = None

modified_at: str | None = None

__init__(id, name=None, queues=None, url=None, content='<omitted>', metadata=None, modified_by=None, modified_at=None)

Update Operations

rossum_mcp.tools.update.handler.register_update_tools(mcp, client, base_url)

Return type:

None

class rossum_mcp.tools.update.models.SchemaNodeUpdate(label=None, type=None, score_threshold=None, hidden=None, disable_prediction=None, can_export=None, default_value=None, rir_field_names=None, constraints=None, options=None, ui_configuration=None, formula=None, prompt=None, context=None, matching=None, enum_value_type=None, width=None, stretch=None, min_occurrences=None, max_occurrences=None)

Bases: object

Partial update for an existing schema node.

Only include fields you want to update - all fields are optional.

label: str | None = None

type: DatapointType | None = None

score_threshold: float | None = None

hidden: bool | None = None

disable_prediction: bool | None = None

can_export: bool | None = None

default_value: str | None = None

rir_field_names: list[str] | None = None

constraints: dict | None = None

options: list[dict] | None = None

ui_configuration: dict | None = None

formula: str | None = None

prompt: str | None = None

context: list[str] | None = None

matching: dict | None = None

enum_value_type: str | None = None

width: int | None = None

stretch: bool | None = None

min_occurrences: int | None = None

max_occurrences: int | None = None

to_dict()

Return type:

dict

__init__(label=None, type=None, score_threshold=None, hidden=None, disable_prediction=None, can_export=None, default_value=None, rir_field_names=None, constraints=None, options=None, ui_configuration=None, formula=None, prompt=None, context=None, matching=None, enum_value_type=None, width=None, stretch=None, min_occurrences=None, max_occurrences=None)

class rossum_mcp.tools.update.models.QueueUpdateData

Bases: TypedDict

name: str

automation_enabled: bool

automation_level: AutomationLevel

locale: QueueLocale

metadata: dict[str, Any]

settings: dict[str, Any]

engine: str

dedicated_engine: str

training_enabled: bool

webhooks: list[str]

hooks: list[str]

default_score_threshold: float

session_timeout: str

document_lifetime: str | None

delete_after: str | None

schema: str

workspace: str

connector: str | None

inbox: str | None

class rossum_mcp.tools.update.models.EngineUpdateData

Bases: TypedDict

name: str

description: str

learning_enabled: bool

training_queues: list[str]

Schema Updates

Schema update operations: update, patch, prune.

Schema patching utilities for Rossum MCP Server.

class rossum_mcp.tools.update.schemas.patching.PatchOperation(*values)

Bases: StrEnum

ADD = 'add'

UPDATE = 'update'

REMOVE = 'remove'

rossum_mcp.tools.update.schemas.patching.apply_schema_patch(content, operation, node_id, node_data=None, parent_id=None, position=None)

Apply a patch operation to schema content.

Return type:

list[dict]

Schema pruning utilities for removing fields from schema content.

Delete Operations

rossum_mcp.tools.delete.handler.register_delete_tools(mcp, client)

Return type:

None

rossum_mcp.tools.delete.registry.build_delete_registry(client)

Return type:

dict[str, Callable[[int], Awaitable[dict]]]

class rossum_mcp.tools.delete.models.DeleteEntityType(*values)

Bases: StrEnum

QUEUE = 'queue'

SCHEMA = 'schema'

HOOK = 'hook'

RULE = 'rule'

WORKSPACE = 'workspace'

ANNOTATION = 'annotation'