Skip to content

BaseTool

BaseTool

Bases: ABC

Abstract base for MCPStack tools with lifecycle hooks and backends.

BaseTool defines the minimal contract every MCP tool must implement so it can be composed into a pipeline and exposed to an MCP host.

Core responsibilities
  • Declare actions (callables) that the MCP server will expose.
  • Manage lifecycle (initialize / teardown / post_load) and any underlying backends (clients, connections, caches).
  • Provide (de-)serialization via to_dict / from_dict.

What is an action?

An action is a Python callable (function or bound method) that the MCP server exposes to the LLM or client. MCPStack registers your actions by calling :meth:actions, then internally doing self.mcp.tool()(action) for each callable.

Keep signatures simple and data JSON-serializable.

Minimal custom tool

from MCPStack.core.tool.base import BaseTool

class HelloTool(BaseTool):
    def __init__(self, greeting: str = "Hello"):
        super().__init__()
        self.greeting = greeting
        self.required_env_vars = {"HELLO_API_KEY": None}  # required

    def actions(self) -> list[callable]:
        # Expose the bound method below as an MCP action
        return [self.say_hello]

    def _initialize(self) -> None:
        # Create clients, read env, warm caches, etc.
        self.api_key = self.required_env_vars.get("HELLO_API_KEY")

    def _teardown(self) -> None:
        # Close clients if needed
        pass

    def _post_load(self) -> None:
        # Reconnect handles after deserialization
        pass

    def say_hello(self, name: str) -> dict:
        '''Return a greeting payload.'''
        return {"message": f"{self.greeting}, {name}!"}

    def to_dict(self) -> dict:
        return {"greeting": self.greeting}

    @classmethod
    def from_dict(cls, params: dict) -> "HelloTool":
        return cls(**params)

Attributes:

Name Type Description
required_env_vars dict[str, Optional[str]]

Names and defaults for env vars the tool needs. A value of None marks a required variable; non-None acts as a default.
backends dict[str, Any]

Optional backing resources (e.g., DB clients). If a backend object implements initialize() and/or teardown(), they will be called automatically.

Backends

A tool can expose multiple backends; each may have initialize() / teardown(). Keep them idempotent — lifecycle hooks may run more than once across builds or reloads.

Serialization boundary

Only configuration/state should be persisted via to_dict. Do not serialize live handles (DB connections, HTTP clients). Recreate those in _initialize or _post_load.

Source code in src/MCPStack/core/tool/base.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
@beartype
class BaseTool(ABC):
    """Abstract base for MCPStack tools with lifecycle hooks and backends.

    `BaseTool` defines the minimal contract every MCP tool must implement so it
    can be composed into a pipeline and exposed to an MCP host.

    Core responsibilities:
      * Declare **actions** (callables) that the MCP server will expose.
      * Manage **lifecycle** (`initialize` / `teardown` / `post_load`) and any
        underlying **backends** (clients, connections, caches).
      * Provide **(de-)serialization** via `to_dict` / `from_dict`.

    !!! tip "What is an *action*?"
        An **action** is a Python callable (function or bound method) that the
        MCP server exposes to the LLM or client. MCPStack registers your
        actions by calling :meth:`actions`, then internally doing
        `self.mcp.tool()(action)` for each callable.

        *Keep signatures simple and data JSON-serializable.*

    !!! example "Minimal custom tool"
        ```python
        from MCPStack.core.tool.base import BaseTool

        class HelloTool(BaseTool):
            def __init__(self, greeting: str = "Hello"):
                super().__init__()
                self.greeting = greeting
                self.required_env_vars = {"HELLO_API_KEY": None}  # required

            def actions(self) -> list[callable]:
                # Expose the bound method below as an MCP action
                return [self.say_hello]

            def _initialize(self) -> None:
                # Create clients, read env, warm caches, etc.
                self.api_key = self.required_env_vars.get("HELLO_API_KEY")

            def _teardown(self) -> None:
                # Close clients if needed
                pass

            def _post_load(self) -> None:
                # Reconnect handles after deserialization
                pass

            def say_hello(self, name: str) -> dict:
                '''Return a greeting payload.'''
                return {"message": f"{self.greeting}, {name}!"}

            def to_dict(self) -> dict:
                return {"greeting": self.greeting}

            @classmethod
            def from_dict(cls, params: dict) -> "HelloTool":
                return cls(**params)
        ```

    Attributes:
        required_env_vars (dict[str, Optional[str]]): Names and defaults for
            env vars the tool needs. A value of `None` marks a **required**
            variable; non-`None` acts as a default.
        backends (dict[str, Any]): Optional backing resources (e.g., DB
            clients). If a backend object implements `initialize()` and/or
            `teardown()`, they will be called automatically.

    !!! note "Backends"
        A tool can expose multiple backends; each may have `initialize()` /
        `teardown()`. Keep them **idempotent** — lifecycle hooks may run more
        than once across builds or reloads.

    !!! warning "Serialization boundary"
        Only **configuration/state** should be persisted via `to_dict`. Do not
        serialize live handles (DB connections, HTTP clients). Recreate those
        in `_initialize` or `_post_load`.
    """

    def __init__(self) -> None:
        self.required_env_vars: Dict[str, Optional[str]] = {}
        self.backends: Dict[str, Any] = {}

    @abstractmethod
    def actions(self) -> list[Callable]:
        """Return the list of callables to be registered as MCP actions.

        MCPStack will iterate this list and register each callable with the MCP
        server. Each callable becomes invokable by the client/LLM.

        !!! tip "Designing an action"
            - Keep parameters and return values **JSON-serializable**.
            - Prefer **explicit, typed** parameters; avoid `*args/**kwargs`.
            - Validate inputs early; raise a descriptive `MCPStackError` on
              misuse.
            - Return **small payloads** or **stream** large data via backends,
              depending on your host capabilities.

        !!! example "Typical return"
            ```python
            def get_schema(self, table: str) -> dict:
                return {"table": table, "columns": [...], "primary_key": "id"}
            ```

        Returns:
            list[Callable]: The MCP-exposed actions in this tool.
        """
        ...

    def initialize(self) -> None:
        """Initialize all backends, then call the tool's `_initialize()` hook.

        Lifecycle entry point invoked by MCPStack during :meth:`build` or
        :meth:`post_load`.

        Steps:
          1. For each backend in :attr:`backends`, call `backend.initialize()`
             if present.
          2. Invoke :meth:`_initialize` for tool-specific setup.

        !!! success "Idempotent by design"
            Your initialization should be safe to call multiple times. Guard
            with flags if needed (e.g., `if self._ready: return`).
        """
        for backend in self.backends.values():
            if hasattr(backend, "initialize"):
                backend.initialize()
        self._initialize()

    def teardown(self) -> None:
        """Run the tool's `_teardown()` then attempt to teardown each backend.

        Called by MCPStack on server shutdown so tools can release resources.

        Steps:
          1. Invoke :meth:`_teardown` for tool-specific cleanup.
          2. For each backend in :attr:`backends`, call `backend.teardown()` if
             present. Errors are logged at DEBUG and **suppressed**.

        !!! warning "Be robust"
            Teardown should never raise fatally — leave the system in a
            consistent state even if some backends fail to close cleanly.
        """
        self._teardown()
        for backend in self.backends.values():
            try:
                if hasattr(backend, "teardown"):
                    backend.teardown()
            except Exception:
                logger.debug("Backend teardown error", exc_info=True)

    def post_load(self) -> None:
        """Hook called after deserialization; re-initializes the tool.

        Invoked by MCPStack after :meth:`from_dict` when a pipeline is loaded
        from disk. Use this to re-bind handles that cannot be serialized.

        Order:
          1. :meth:`_post_load`
          2. :meth:`initialize`

        !!! tip "What belongs here?"
            - Re-construct in-memory caches.
            - Recreate clients/sessions that depend on current process env.
        """
        self._post_load()
        self.initialize()

    def _initialize(self) -> None:
        """Optional hook for subclasses to implement `initialize` logic.

        Implement tool-specific setup here (e.g., reading env vars, creating
        clients, warming caches).

        !!! example
            ```python
            def _initialize(self) -> None:
                self.api_key = os.getenv("MY_API_KEY")
                self.client = Client(self.api_key)
            ```
        """
        ...

    def _teardown(self) -> None:
        """Optional hook for subclasses to implement `teardown` logic.

        Implement tool-specific cleanup here (close cursors, flush buffers,
        persist metrics, etc.).
        """
        ...

    def _post_load(self) -> None:
        """Optional hook for subclasses to implement `post_load` logic.

        Called immediately after deserialization but before :meth:`initialize`.
        Use this to restore transient state that isn't persisted by `to_dict`.
        """
        ...

    @abstractmethod
    def to_dict(self) -> Dict[str, Any]:
        """Serialize this tool's configuration to a JSON-serializable mapping.

        Only include **configuration** — not live connections. The output of
        this method must be consumable by :meth:`from_dict`.

        Returns:
            Dict[str, Any]: Tool parameters and metadata.

        !!! example
            ```python
            def to_dict(self) -> dict:
                return {"endpoint": self.endpoint, "timeout": self.timeout}
            ```
        """
        ...

    @classmethod
    @abstractmethod
    def from_dict(cls, params: Dict[str, Any]):
        """Create a tool instance from a serialized mapping.

        This is the inverse of :meth:`to_dict`. Construct the tool using the
        provided parameters but **do not** create live connections here — do
        that in `_initialize` or `_post_load`.

        Args:
            params: Mapping produced by `to_dict()`.

        Returns:
            BaseTool: A configured tool instance.

        !!! example
            ```python
            @classmethod
            def from_dict(cls, params: dict) -> "MyTool":
                return cls(**params)
            ```
        """
        ...

actions() abstractmethod

Return the list of callables to be registered as MCP actions.

MCPStack will iterate this list and register each callable with the MCP server. Each callable becomes invokable by the client/LLM.

Designing an action

  • Keep parameters and return values JSON-serializable.
  • Prefer explicit, typed parameters; avoid *args/**kwargs.
  • Validate inputs early; raise a descriptive MCPStackError on misuse.
  • Return small payloads or stream large data via backends, depending on your host capabilities.

Typical return

def get_schema(self, table: str) -> dict:
    return {"table": table, "columns": [...], "primary_key": "id"}

Returns:

Type Description
list[Callable]

list[Callable]: The MCP-exposed actions in this tool.
Source code in src/MCPStack/core/tool/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@abstractmethod
def actions(self) -> list[Callable]:
    """Return the list of callables to be registered as MCP actions.

    MCPStack will iterate this list and register each callable with the MCP
    server. Each callable becomes invokable by the client/LLM.

    !!! tip "Designing an action"
        - Keep parameters and return values **JSON-serializable**.
        - Prefer **explicit, typed** parameters; avoid `*args/**kwargs`.
        - Validate inputs early; raise a descriptive `MCPStackError` on
          misuse.
        - Return **small payloads** or **stream** large data via backends,
          depending on your host capabilities.

    !!! example "Typical return"
        ```python
        def get_schema(self, table: str) -> dict:
            return {"table": table, "columns": [...], "primary_key": "id"}
        ```

    Returns:
        list[Callable]: The MCP-exposed actions in this tool.
    """
    ...

from_dict(params) abstractmethod classmethod

Create a tool instance from a serialized mapping.

This is the inverse of :meth:to_dict. Construct the tool using the provided parameters but do not create live connections here — do that in _initialize or _post_load.

Parameters:

Name Type Description Default
params Dict[str, Any]

Mapping produced by to_dict().

 required

Returns:

Name Type Description
BaseTool

A configured tool instance.

Example

@classmethod
def from_dict(cls, params: dict) -> "MyTool":
    return cls(**params)
Source code in src/MCPStack/core/tool/base.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
@classmethod
@abstractmethod
def from_dict(cls, params: Dict[str, Any]):
    """Create a tool instance from a serialized mapping.

    This is the inverse of :meth:`to_dict`. Construct the tool using the
    provided parameters but **do not** create live connections here — do
    that in `_initialize` or `_post_load`.

    Args:
        params: Mapping produced by `to_dict()`.

    Returns:
        BaseTool: A configured tool instance.

    !!! example
        ```python
        @classmethod
        def from_dict(cls, params: dict) -> "MyTool":
            return cls(**params)
        ```
    """
    ...

initialize()

Initialize all backends, then call the tool's _initialize() hook.

Lifecycle entry point invoked by MCPStack during :meth:build or :meth:post_load.

Steps
  1. For each backend in :attr:backends, call backend.initialize() if present.
  2. Invoke :meth:_initialize for tool-specific setup.

Idempotent by design

Your initialization should be safe to call multiple times. Guard with flags if needed (e.g., if self._ready: return).

Source code in src/MCPStack/core/tool/base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def initialize(self) -> None:
    """Initialize all backends, then call the tool's `_initialize()` hook.

    Lifecycle entry point invoked by MCPStack during :meth:`build` or
    :meth:`post_load`.

    Steps:
      1. For each backend in :attr:`backends`, call `backend.initialize()`
         if present.
      2. Invoke :meth:`_initialize` for tool-specific setup.

    !!! success "Idempotent by design"
        Your initialization should be safe to call multiple times. Guard
        with flags if needed (e.g., `if self._ready: return`).
    """
    for backend in self.backends.values():
        if hasattr(backend, "initialize"):
            backend.initialize()
    self._initialize()

post_load()

Hook called after deserialization; re-initializes the tool.

Invoked by MCPStack after :meth:from_dict when a pipeline is loaded from disk. Use this to re-bind handles that cannot be serialized.

Order
  1. :meth:_post_load
  2. :meth:initialize

What belongs here?

  • Re-construct in-memory caches.
  • Recreate clients/sessions that depend on current process env.
Source code in src/MCPStack/core/tool/base.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def post_load(self) -> None:
    """Hook called after deserialization; re-initializes the tool.

    Invoked by MCPStack after :meth:`from_dict` when a pipeline is loaded
    from disk. Use this to re-bind handles that cannot be serialized.

    Order:
      1. :meth:`_post_load`
      2. :meth:`initialize`

    !!! tip "What belongs here?"
        - Re-construct in-memory caches.
        - Recreate clients/sessions that depend on current process env.
    """
    self._post_load()
    self.initialize()

teardown()

Run the tool's _teardown() then attempt to teardown each backend.

Called by MCPStack on server shutdown so tools can release resources.

Steps
  1. Invoke :meth:_teardown for tool-specific cleanup.
  2. For each backend in :attr:backends, call backend.teardown() if present. Errors are logged at DEBUG and suppressed.

Be robust

Teardown should never raise fatally — leave the system in a consistent state even if some backends fail to close cleanly.

Source code in src/MCPStack/core/tool/base.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def teardown(self) -> None:
    """Run the tool's `_teardown()` then attempt to teardown each backend.

    Called by MCPStack on server shutdown so tools can release resources.

    Steps:
      1. Invoke :meth:`_teardown` for tool-specific cleanup.
      2. For each backend in :attr:`backends`, call `backend.teardown()` if
         present. Errors are logged at DEBUG and **suppressed**.

    !!! warning "Be robust"
        Teardown should never raise fatally — leave the system in a
        consistent state even if some backends fail to close cleanly.
    """
    self._teardown()
    for backend in self.backends.values():
        try:
            if hasattr(backend, "teardown"):
                backend.teardown()
        except Exception:
            logger.debug("Backend teardown error", exc_info=True)

to_dict() abstractmethod

Serialize this tool's configuration to a JSON-serializable mapping.

Only include configuration — not live connections. The output of this method must be consumable by :meth:from_dict.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: Tool parameters and metadata.

Example

def to_dict(self) -> dict:
    return {"endpoint": self.endpoint, "timeout": self.timeout}
Source code in src/MCPStack/core/tool/base.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@abstractmethod
def to_dict(self) -> Dict[str, Any]:
    """Serialize this tool's configuration to a JSON-serializable mapping.

    Only include **configuration** — not live connections. The output of
    this method must be consumable by :meth:`from_dict`.

    Returns:
        Dict[str, Any]: Tool parameters and metadata.

    !!! example
        ```python
        def to_dict(self) -> dict:
            return {"endpoint": self.endpoint, "timeout": self.timeout}
        ```
    """
    ...