Skip to content

teff.tool

teff.tool

Modules:

Name Description
agent

Sub-agent tools: a Tool that drives a short ReAct loop.
builtin
mcp

Model Context Protocol (MCP) bridge and ready-made server presets.
registry

Tool registry and decorator.
tool

Abstract base for all tools.

Classes:

Name Description
AgentTool

A tool that runs a sub-agent ReAct loop over a slice of the tool set.
McpPreset

Base class for launch configs of known MCP servers.
McpTool

A :class:~teff.tool.Tool that forwards calls to an MCP server.
McpToolGroup

A lazily-opened MCP server connection, exposing its tools.
Tool

Abstract base class for tools callable by nodes.
ToolRegistry

Registry mapping tool names to tool classes.

Functions:

Name Description
mcp_tools

Connect to an MCP server and yield its tools as Teff :class:Tool\s.
open_tools

Expand :class:McpToolGroup entries in tools into their members.

AgentTool

Bases: Tool

A tool that runs a sub-agent ReAct loop over a slice of the tool set.

Subclasses declare content and leave the LLM plumbing to this class:

  • system — the sub-agent system message (a constant, or override :meth:system_prompt for state-dependent text);
  • tools — the slice of the tool set the sub-agent may call, as a mapping {name: Tool} or a plain iterable of :class:Tool instances (keyed by tool.name);
  • user_message — a template with {placeholder} fields resolved against state: {state_key} takes the value of that key, and derived values (e.g. a schema-rendered project_info) go through the formatters mapping; override :meth:user_message for fully custom messages;
  • :meth:handle_reply — turn the final reply into the tool result and, optionally, write results back into state.

For the common shapes the constructor takes: writes=("plan",) copies the raw reply text into those state keys (with :meth:handle_reply returning the text), and :meth:json_reply parses a JSON reply against a schema.

The runtime injects __state__ / __ctx__ (see :func:teff.harness.tools); anything :meth:handle_reply writes into state is copied back into the enclosing workflow state.

Methods:

Name Description
arun

Run the sub-agent and surface the handled reply as the result.
handle_reply

Post-process the final reply into the tool result.
json_reply

Parse the final reply as a JSON object matching schema.
system_prompt

Return the sub-agent's system message (may read state).
user_message

Return the sub-agent's user message (may read state).
Source code in teff/tool/agent.py 
 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
class AgentTool(Tool):
    """A tool that runs a sub-agent ReAct loop over a slice of the tool set.

    Subclasses declare *content* and leave the LLM plumbing to this class:

    - ``system`` — the sub-agent system message (a constant, or override
      :meth:`system_prompt` for state-dependent text);
    - ``tools`` — the slice of the tool set the sub-agent may call, as a
      mapping ``{name: Tool}`` or a plain iterable of :class:`Tool` instances
      (keyed by ``tool.name``);
    - ``user_message`` — a template with ``{placeholder}`` fields resolved
      against *state*: ``{state_key}`` takes the value of that key, and
      derived values (e.g. a schema-rendered ``project_info``) go through
      the ``formatters`` mapping; override :meth:`user_message` for fully
      custom messages;
    - :meth:`handle_reply` — turn the final reply into the tool result and,
      optionally, write results back into *state*.

    For the common shapes the constructor takes:
    ``writes=("plan",)`` copies the raw reply text into those state keys
    (with :meth:`handle_reply` returning the text), and :meth:`json_reply`
    parses a JSON reply against a schema.

    The runtime injects ``__state__`` / ``__ctx__`` (see
    :func:`teff.harness.tools`); anything :meth:`handle_reply` writes into
    *state* is copied back into the enclosing workflow state.
    """

    #: Class-attribute defaults, overridable per instance via the constructor.
    #: Declare these on a subclass to describe a sub-agent without an
    #: ``__init__`` override.
    system: str = ""
    max_rounds: int = 10
    writes: tuple[str, ...] = ()
    user_template: str = ""
    formatters: "dict[str, Callable[[dict], str]] | None" = None

    def __init__(
        self,
        model: str,
        provider: str,
        *,
        tools: "Mapping[str, Tool] | Iterable[Tool] | None" = None,
        system: str | None = None,
        max_rounds: int | None = None,
        writes: tuple[str, ...] | None = None,
        user_template: str | None = None,
        formatters: "dict[str, Callable[[dict], str]] | None" = None,
    ):
        super().__init__()
        self._model = model
        self._provider = provider
        self._tools = self._tool_map(tools)
        self._system = system if system is not None else self.system
        self._max_rounds = max_rounds if max_rounds is not None else self.max_rounds
        self._writes = tuple(writes if writes is not None else self.writes)
        self._user_template = (
            user_template if user_template is not None else self.user_template
        )
        self._formatters = dict(
            formatters if formatters is not None else (self.formatters or {})
        )

    # -- subclass extension points ------------------------------------------
    def system_prompt(self, state: dict) -> str:
        """Return the sub-agent's system message (may read *state*)."""
        return self._system

    def user_message(self, state: dict) -> str:
        """Return the sub-agent's user message (may read *state*).

        Default: render the ``user_template`` constructor param, resolving
        ``{name}`` fields via ``formatters[name](state)`` when registered,
        else from ``state[name]`` (empty when absent).  Override for fully
        custom messages that don't fit the template.
        """
        if not self._user_template:
            return ""
        return self._render_user(state)

    def _render_user(self, state: dict) -> str:
        names = {
            field for _, field, _, _ in Formatter().parse(self._user_template) if field
        }
        values: dict[str, str] = {}
        for name in names:
            if name in self._formatters:
                values[name] = str(self._formatters[name](state))
            elif name in state:
                values[name] = str(state.get(name) or "")
            else:
                raise ValueError(
                    f"AgentTool '{self.name}': no formatter and no state key for "
                    f"placeholder '{{{name}}}' in user_template"
                )
        return self._user_template.format(**values)

    def handle_reply(self, state: dict, reply) -> str:
        """Post-process the final reply into the tool result.

        May read and write *state*; writes are copied back into the
        enclosing workflow state.  ``reply`` is the final
        :class:`~teff.harness.loop.Step` (``reply.content`` holds the text).

        Default: copy the reply text into each ``writes`` key and return it.
        """
        content = (reply.content or "").strip()
        for key in self._writes:
            state[key] = content
        return content

    def json_reply(self, reply, schema: dict) -> dict | None:
        """Parse the final reply as a JSON object matching *schema*.

        Returns the parsed dict, or ``None`` when the reply is not valid
        JSON (or fails validation) — failures fall back to ``None`` rather
        than raising, mirroring how structured outputs are best-effort.
        """
        try:
            parsed = parse_json_object(reply.content or "")
            if isinstance(parsed, dict) and not validate_json(parsed, schema):
                return parsed
        except ValueError:
            pass
        return None

    @staticmethod
    def _tool_map(tools: "Mapping[str, Tool] | Iterable[Tool] | None") -> dict:
        """Normalize ``tools`` to a ``{name: Tool}`` map.

        Accepts either a mapping or a plain iterable of :class:`Tool`
        instances, keyed by each tool's ``name`` attribute.
        """
        if not tools:
            return {}
        if isinstance(tools, Mapping):
            return dict(tools)
        return {t.name: t for t in tools}

    # -- framework plumbing --------------------------------------------------
    def _harness(self, ctx):
        """Build a sub-agent harness wired to the active exec context."""
        from teff.harness import Harness  # local import breaks the tool↔harness cycle

        harness = Harness.from_config(
            {
                "model": self._model,
                "provider": self._provider,
                "max_tool_rounds": self._max_rounds,
                "parse_text_tool_calls": True,
            },
            default_provider=self._provider,
            default_model=self._model,
            providers=getattr(ctx, "providers", None) if ctx is not None else None,
        )
        if ctx is not None:
            tracer = getattr(ctx, "tracer", None)
            if tracer is not None:

                async def on_llm(provider, model, prompt, completion, duration):
                    tracer.llm(provider, model, prompt, completion, duration)

                harness.on_llm = on_llm
            if getattr(ctx, "on_llm_payload", None) is not None:
                harness.on_llm_payload = ctx.on_llm_payload
        return harness

    async def arun(self, *, __state__=None, __ctx__=None, **kwargs):  # noqa: ARG002
        """Run the sub-agent and surface the handled reply as the result."""
        state = dict(__state__ or {})
        messages = [
            {"role": "system", "content": self.system_prompt(state)},
            {"role": "user", "content": self.user_message(state)},
        ]
        reply = await self._harness(__ctx__).run(messages, tools=self._tools or None)
        result = self.handle_reply(state, reply)
        if __state__ is not None:
            __state__.update(state)
        return result

arun async 

arun(*, __state__=None, __ctx__=None, **kwargs)

Run the sub-agent and surface the handled reply as the result.

Source code in teff/tool/agent.py 
193
194
195
196
197
198
199
200
201
202
203
204
async def arun(self, *, __state__=None, __ctx__=None, **kwargs):  # noqa: ARG002
    """Run the sub-agent and surface the handled reply as the result."""
    state = dict(__state__ or {})
    messages = [
        {"role": "system", "content": self.system_prompt(state)},
        {"role": "user", "content": self.user_message(state)},
    ]
    reply = await self._harness(__ctx__).run(messages, tools=self._tools or None)
    result = self.handle_reply(state, reply)
    if __state__ is not None:
        __state__.update(state)
    return result

handle_reply 

handle_reply(state, reply)

Post-process the final reply into the tool result.

May read and write state; writes are copied back into the enclosing workflow state. reply is the final :class:~teff.harness.loop.Step (reply.content holds the text).

Default: copy the reply text into each writes key and return it.

Source code in teff/tool/agent.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
def handle_reply(self, state: dict, reply) -> str:
    """Post-process the final reply into the tool result.

    May read and write *state*; writes are copied back into the
    enclosing workflow state.  ``reply`` is the final
    :class:`~teff.harness.loop.Step` (``reply.content`` holds the text).

    Default: copy the reply text into each ``writes`` key and return it.
    """
    content = (reply.content or "").strip()
    for key in self._writes:
        state[key] = content
    return content

json_reply 

json_reply(reply, schema)

Parse the final reply as a JSON object matching schema.

Returns the parsed dict, or None when the reply is not valid JSON (or fails validation) — failures fall back to None rather than raising, mirroring how structured outputs are best-effort.

Source code in teff/tool/agent.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def json_reply(self, reply, schema: dict) -> dict | None:
    """Parse the final reply as a JSON object matching *schema*.

    Returns the parsed dict, or ``None`` when the reply is not valid
    JSON (or fails validation) — failures fall back to ``None`` rather
    than raising, mirroring how structured outputs are best-effort.
    """
    try:
        parsed = parse_json_object(reply.content or "")
        if isinstance(parsed, dict) and not validate_json(parsed, schema):
            return parsed
    except ValueError:
        pass
    return None

system_prompt 

system_prompt(state)

Return the sub-agent's system message (may read state).

Source code in teff/tool/agent.py 
90
91
92
def system_prompt(self, state: dict) -> str:
    """Return the sub-agent's system message (may read *state*)."""
    return self._system

user_message 

user_message(state)

Return the sub-agent's user message (may read state).

Default: render the user_template constructor param, resolving {name} fields via formatters[name](state) when registered, else from state[name] (empty when absent). Override for fully custom messages that don't fit the template.

Source code in teff/tool/agent.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def user_message(self, state: dict) -> str:
    """Return the sub-agent's user message (may read *state*).

    Default: render the ``user_template`` constructor param, resolving
    ``{name}`` fields via ``formatters[name](state)`` when registered,
    else from ``state[name]`` (empty when absent).  Override for fully
    custom messages that don't fit the template.
    """
    if not self._user_template:
        return ""
    return self._render_user(state)

McpPreset

Base class for launch configs of known MCP servers.

Subclasses declare the canonical registry name plus either the stdio command (or streamable-http url) and the env keys the server expects. env values are overridable defaults; from_preset merges any caller-provided env over them (same key wins), and an explicit command/url fully replaces the preset's transport.

Source code in teff/tool/mcp/presets.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class McpPreset:
    """Base class for launch configs of known MCP servers.

    Subclasses declare the canonical registry ``name`` plus either the
    stdio ``command`` (or streamable-http ``url``) and the ``env`` keys the
    server expects.  ``env`` values are overridable defaults;
    ``from_preset`` merges any caller-provided env over them (same key
    wins), and an explicit ``command``/``url`` fully replaces the preset's
    transport.
    """

    #: Canonical name used for ``preset:``/``from_preset``.
    name: ClassVar[str] = ""
    #: Stdio launch command (or set ``url`` instead).
    command: ClassVar[list[str] | None] = None
    #: Streamable-http endpoint (or set ``command`` instead).
    url: ClassVar[str | None] = None
    #: Env-var keys the server expects; values act as defaults.
    env: ClassVar[dict[str, str]] = {}
    #: One-line description of the server.
    description: ClassVar[str | None] = None

McpTool

Bases: Tool

A :class:~teff.tool.Tool that forwards calls to an MCP server.

Instances are created by :func:mcp_tools. The tool's JSON schema comes from the server's tool definition instead of being inferred from type hints.

Source code in teff/tool/mcp/bridge.py 
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
class McpTool(Tool):
    """A :class:`~teff.tool.Tool` that forwards calls to an MCP server.

    Instances are created by :func:`mcp_tools`.  The tool's JSON schema
    comes from the server's tool definition instead of being inferred
    from type hints.
    """

    def __init__(self, session: "ClientSession", spec: "McpToolSpec"):
        super().__init__()
        self._session = session
        self._server_name = spec.name
        self.name = spec.name
        self.description = spec.description or ""
        # SDK stubs expose `inputSchema`, runtime uses `input_schema`.
        self.schema = spec.input_schema  # type: ignore[attr-defined]

    async def arun(self, **kwargs):
        result = await self._session.call_tool(self._server_name, kwargs)
        content = _format_mcp_content(result)
        if getattr(result, "is_error", False):
            raise RuntimeError(
                content or f"MCP tool '{self._server_name}' returned an error"
            )
        return content

McpToolGroup

A lazily-opened MCP server connection, exposing its tools.

Holds the connection config (url or command, env, cwd) without any live connection. The session is opened on first :meth:open and kept open until :meth:aclose, so several graph.run calls (daemon ticks, conversation turns, resumes) share a single connection instead of re-spawning the server each time.

Instances are created from a workflow's tools: block (type: mcp) or directly::

group = McpToolGroup(id="drive", command=["uvx", "mcp-server-google-drive"])
tools = await group.open()      # -> [McpTool, ...] named ``drive__<tool>``
...
await group.aclose()

Ready-made presets for known servers give the launch command and its env-var keys in one shot; overrides merge on top::

group = McpToolGroup.from_preset(
    "google_drive",
    env={"GOOGLE_DRIVE_REFRESH_TOKEN": os.environ["GDRIVE"]},
)

Parameters:

Name Type Description Default
id str

Server id; member tools are prefixed <id>__<name> so tools from different servers never collide.

 'mcp'
url str | None

Streamable HTTP endpoint (mutually exclusive with command).

 None
command list[str] | None

Stdio server command, a list of argv tokens.

 None
env dict[str, str] | None

Optional extra environment variables for stdio servers.

 None
cwd str | None

Optional working directory for stdio servers.

 None
client_info dict | None

Optional dict overrides for the client Implementation advertised to the server.

 None

Methods:

Name Description
aclose

Close the connection if it was opened. Idempotent.
from_preset

Build a group from a named :data:MCP_PRESETS entry.
open

Return the server's tools, opening the connection on first use.
Source code in teff/tool/mcp/bridge.py 
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
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
class McpToolGroup:
    """A lazily-opened MCP server connection, exposing its tools.

    Holds the connection *config* (url or command, env, cwd) without any
    live connection.  The session is opened on first :meth:`open` and kept
    open until :meth:`aclose`, so several ``graph.run`` calls (daemon ticks,
    conversation turns, resumes) share a single connection instead of
    re-spawning the server each time.

    Instances are created from a workflow's ``tools:`` block (``type: mcp``)
    or directly::

        group = McpToolGroup(id="drive", command=["uvx", "mcp-server-google-drive"])
        tools = await group.open()      # -> [McpTool, ...] named ``drive__<tool>``
        ...
        await group.aclose()

    Ready-made presets for known servers give the launch command and its
    env-var keys in one shot; overrides merge on top::

        group = McpToolGroup.from_preset(
            "google_drive",
            env={"GOOGLE_DRIVE_REFRESH_TOKEN": os.environ["GDRIVE"]},
        )

    Args:
        id: Server id; member tools are prefixed ``<id>__<name>`` so tools
            from different servers never collide.
        url: Streamable HTTP endpoint (mutually exclusive with *command*).
        command: Stdio server command, a list of argv tokens.
        env: Optional extra environment variables for stdio servers.
        cwd: Optional working directory for stdio servers.
        client_info: Optional dict overrides for the client
            ``Implementation`` advertised to the server.
    """

    def __init__(
        self,
        *,
        id: str = "mcp",
        url: str | None = None,
        command: list[str] | None = None,
        env: dict[str, str] | None = None,
        cwd: str | None = None,
        client_info: dict | None = None,
    ):
        if (url is None) == (command is None):
            raise ValueError("McpToolGroup requires exactly one of 'url' or 'command'")
        self.id = id
        self._url = url
        self._command = command
        self._env = env
        self._cwd = cwd
        self._client_info = client_info
        self._stack: contextlib.AsyncExitStack | None = None
        self._tools: list[McpTool] | None = None
        self.is_mcp_group = self

    @classmethod
    def from_preset(
        cls,
        name: str,
        *,
        env: dict[str, str] | None = None,
        **overrides,
    ) -> "McpToolGroup":
        """Build a group from a named :data:`MCP_PRESETS` entry.

        The preset class supplies its canonical ``name`` (used as the
        default ``id``), the launch ``command`` (or ``url``) and its default
        ``env`` keys.  *env* entries merge over the preset's defaults (same
        key overrides); ``command``/``url``/``id``/``cwd`` overrides fully
        replace the preset's value.

        Raises:
            KeyError: If *name* is not a known preset.
        """
        preset: type[McpPreset] = _lookup_preset(name)
        if (preset.url is None) == (preset.command is None):
            raise ValueError(
                f"MCP preset {name!r} must define exactly one of 'url' or 'command'"
            )
        cfg: dict = {"id": preset.name}
        if preset.url is not None:
            cfg["url"] = preset.url
        else:
            cfg["command"] = preset.command
        if preset.env:
            cfg["env"] = dict(preset.env)
        cfg.update(overrides)
        merged_env = {**(cfg.get("env") or {}), **(env or {})}
        if merged_env:
            cfg["env"] = merged_env
        return cls(**cfg)

    async def open(self) -> list[McpTool]:
        """Return the server's tools, opening the connection on first use.

        Repeated calls return the cached member tools; the underlying
        session stays open until :meth:`aclose`.
        """
        if self._tools is not None:
            return self._tools
        from mcp import StdioServerParameters
        from mcp.client.stdio import stdio_client
        from mcp.client.streamable_http import streamable_http_client

        stack = contextlib.AsyncExitStack()
        try:
            if self._url is not None:
                read_stream, write_stream = await stack.enter_async_context(  # type: ignore[misc]
                    streamable_http_client(self._url)
                )
            else:
                assert self._command is not None
                _ensure_runtime(self._command[0], self.id)
                params = StdioServerParameters(
                    command=self._command[0],
                    args=self._command[1:],
                    env=self._env,
                    cwd=self._cwd,
                )
                read_stream, write_stream = await stack.enter_async_context(
                    stdio_client(params)
                )
            session, member_tools = await _connect_tools(
                read_stream, write_stream, self._client_info
            )
        except Exception:
            await stack.aclose()
            raise
        stack.push_async_callback(partial(session.__aexit__, None, None, None))
        self._stack = stack
        for tool in member_tools:
            tool.name = f"{self.id}__{tool.name}"
        self._tools = member_tools
        return self._tools

    async def aclose(self) -> None:
        """Close the connection if it was opened.  Idempotent."""
        if self._stack is not None:
            await self._stack.aclose()
            self._stack = None
            self._tools = None

aclose async 

aclose()

Close the connection if it was opened. Idempotent.

Source code in teff/tool/mcp/bridge.py 
265
266
267
268
269
270
async def aclose(self) -> None:
    """Close the connection if it was opened.  Idempotent."""
    if self._stack is not None:
        await self._stack.aclose()
        self._stack = None
        self._tools = None

from_preset classmethod 

from_preset(name, *, env=None, **overrides)

Build a group from a named :data:MCP_PRESETS entry.

The preset class supplies its canonical name (used as the default id), the launch command (or url) and its default env keys. env entries merge over the preset's defaults (same key overrides); command/url/id/cwd overrides fully replace the preset's value.

Raises:

Type Description
KeyError

If name is not a known preset.
Source code in teff/tool/mcp/bridge.py 
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
@classmethod
def from_preset(
    cls,
    name: str,
    *,
    env: dict[str, str] | None = None,
    **overrides,
) -> "McpToolGroup":
    """Build a group from a named :data:`MCP_PRESETS` entry.

    The preset class supplies its canonical ``name`` (used as the
    default ``id``), the launch ``command`` (or ``url``) and its default
    ``env`` keys.  *env* entries merge over the preset's defaults (same
    key overrides); ``command``/``url``/``id``/``cwd`` overrides fully
    replace the preset's value.

    Raises:
        KeyError: If *name* is not a known preset.
    """
    preset: type[McpPreset] = _lookup_preset(name)
    if (preset.url is None) == (preset.command is None):
        raise ValueError(
            f"MCP preset {name!r} must define exactly one of 'url' or 'command'"
        )
    cfg: dict = {"id": preset.name}
    if preset.url is not None:
        cfg["url"] = preset.url
    else:
        cfg["command"] = preset.command
    if preset.env:
        cfg["env"] = dict(preset.env)
    cfg.update(overrides)
    merged_env = {**(cfg.get("env") or {}), **(env or {})}
    if merged_env:
        cfg["env"] = merged_env
    return cls(**cfg)

open async 

open()

Return the server's tools, opening the connection on first use.

Repeated calls return the cached member tools; the underlying session stays open until :meth:aclose.

Source code in teff/tool/mcp/bridge.py 
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
250
251
252
253
254
255
256
257
258
259
260
261
262
263
async def open(self) -> list[McpTool]:
    """Return the server's tools, opening the connection on first use.

    Repeated calls return the cached member tools; the underlying
    session stays open until :meth:`aclose`.
    """
    if self._tools is not None:
        return self._tools
    from mcp import StdioServerParameters
    from mcp.client.stdio import stdio_client
    from mcp.client.streamable_http import streamable_http_client

    stack = contextlib.AsyncExitStack()
    try:
        if self._url is not None:
            read_stream, write_stream = await stack.enter_async_context(  # type: ignore[misc]
                streamable_http_client(self._url)
            )
        else:
            assert self._command is not None
            _ensure_runtime(self._command[0], self.id)
            params = StdioServerParameters(
                command=self._command[0],
                args=self._command[1:],
                env=self._env,
                cwd=self._cwd,
            )
            read_stream, write_stream = await stack.enter_async_context(
                stdio_client(params)
            )
        session, member_tools = await _connect_tools(
            read_stream, write_stream, self._client_info
        )
    except Exception:
        await stack.aclose()
        raise
    stack.push_async_callback(partial(session.__aexit__, None, None, None))
    self._stack = stack
    for tool in member_tools:
        tool.name = f"{self.id}__{tool.name}"
    self._tools = member_tools
    return self._tools

Tool

Abstract base class for tools callable by nodes.

Subclasses define name and description as class attributes, then implement run (sync) and/or arun (async).

Attributes:

Name Type Description
name str

Unique tool name (defaults to lowercase class name).
description str

Human-readable description for LLM tool selection.
schema dict | None

Optional JSON Schema dict for the tool's arguments. When set (e.g. by :class:~teff.tool.mcp.McpTool), it is used as-is instead of being inferred from the run/arun signature.

Methods:

Name Description
arun

Execute the tool asynchronously.
run

Execute the tool synchronously.
Source code in teff/tool/tool.py 
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
class Tool:
    """Abstract base class for tools callable by nodes.

    Subclasses define *name* and *description* as class attributes,
    then implement *run* (sync) and/or *arun* (async).

    Attributes:
        name: Unique tool name (defaults to lowercase class name).
        description: Human-readable description for LLM tool selection.
        schema: Optional JSON Schema dict for the tool's arguments.  When
            set (e.g. by :class:`~teff.tool.mcp.McpTool`), it is used as-is
            instead of being inferred from the ``run``/``arun`` signature.
    """

    name: str = ""
    description: str = ""
    schema: dict | None = None

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        if cls.name == "":
            cls.name = cls.__name__.lower()

    def __init__(self):
        pass

    def run(self, **kwargs):
        """Execute the tool synchronously.

        Args:
            **kwargs: Tool-specific keyword arguments.

        Returns:
            Tool-specific result (typically a string).
        """
        raise NotImplementedError

    async def arun(self, **kwargs):
        """Execute the tool asynchronously.

        Falls back to *run* via ``asyncio.to_thread`` if not overridden.

        Args:
            **kwargs: Tool-specific keyword arguments.

        Returns:
            Tool-specific result (typically a string).
        """
        return await asyncio.to_thread(self.run, **kwargs)

arun async 

arun(**kwargs)

Execute the tool asynchronously.

Falls back to run via asyncio.to_thread if not overridden.

Parameters:

Name Type Description Default
**kwargs

Tool-specific keyword arguments.

 {}

Returns:

Type Description

Tool-specific result (typically a string).
Source code in teff/tool/tool.py 
84
85
86
87
88
89
90
91
92
93
94
95
async def arun(self, **kwargs):
    """Execute the tool asynchronously.

    Falls back to *run* via ``asyncio.to_thread`` if not overridden.

    Args:
        **kwargs: Tool-specific keyword arguments.

    Returns:
        Tool-specific result (typically a string).
    """
    return await asyncio.to_thread(self.run, **kwargs)

run 

run(**kwargs)

Execute the tool synchronously.

Parameters:

Name Type Description Default
**kwargs

Tool-specific keyword arguments.

 {}

Returns:

Type Description

Tool-specific result (typically a string).
Source code in teff/tool/tool.py 
73
74
75
76
77
78
79
80
81
82
def run(self, **kwargs):
    """Execute the tool synchronously.

    Args:
        **kwargs: Tool-specific keyword arguments.

    Returns:
        Tool-specific result (typically a string).
    """
    raise NotImplementedError

ToolRegistry

Registry mapping tool names to tool classes.

Methods:

Name Description
create

Instantiate a tool by name.
list

Return all registered tool names.
register

Register a tool class.
Source code in teff/tool/registry.py 
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
class ToolRegistry:
    """Registry mapping tool names to tool classes."""

    def __init__(self) -> None:
        self._tools: dict[str, type[Tool]] = {}

    def register(self, tool_cls: type[Tool]) -> None:
        """Register a tool class.

        Raises:
            ValueError: If the class has no non-empty *name* attribute.
        """
        if not hasattr(tool_cls, "name") or not tool_cls.name:
            raise ValueError(
                f"Tool class {tool_cls.__name__} must have a non-empty 'name' attribute"
            )
        self._tools[tool_cls.name] = tool_cls

    def create(self, name: str, config: dict | None = None) -> Tool:
        """Instantiate a tool by name.

        If *config* is provided, it is passed to the tool's constructor as a
        dict when the constructor accepts one (a leading ``config``
        parameter); otherwise the config keys are passed as keyword
        arguments, and if the constructor rejects them the values are
        assigned as attributes on an argument-less instance.

        Raises:
            KeyError: If the name is not registered.
        """
        if name not in self._tools:
            msg = f"unknown tool: {name}"
            raise KeyError(msg)
        cls = self._tools[name]
        if config is None:
            return cls()
        if _accepts_config_dict(cls):
            return cls(config)  # type: ignore[call-arg]
        try:
            return cls(**config)
        except TypeError:
            tool = cls()
            for k, v in config.items():
                setattr(tool, k, v)
            return tool

    def list(self) -> list[str]:
        """Return all registered tool names."""
        return list(self._tools.keys())

create 

create(name, config=None)

Instantiate a tool by name.

If config is provided, it is passed to the tool's constructor as a dict when the constructor accepts one (a leading config parameter); otherwise the config keys are passed as keyword arguments, and if the constructor rejects them the values are assigned as attributes on an argument-less instance.

Raises:

Type Description
KeyError

If the name is not registered.
Source code in teff/tool/registry.py 
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
def create(self, name: str, config: dict | None = None) -> Tool:
    """Instantiate a tool by name.

    If *config* is provided, it is passed to the tool's constructor as a
    dict when the constructor accepts one (a leading ``config``
    parameter); otherwise the config keys are passed as keyword
    arguments, and if the constructor rejects them the values are
    assigned as attributes on an argument-less instance.

    Raises:
        KeyError: If the name is not registered.
    """
    if name not in self._tools:
        msg = f"unknown tool: {name}"
        raise KeyError(msg)
    cls = self._tools[name]
    if config is None:
        return cls()
    if _accepts_config_dict(cls):
        return cls(config)  # type: ignore[call-arg]
    try:
        return cls(**config)
    except TypeError:
        tool = cls()
        for k, v in config.items():
            setattr(tool, k, v)
        return tool

list 

list()

Return all registered tool names.

Source code in teff/tool/registry.py 
69
70
71
def list(self) -> list[str]:
    """Return all registered tool names."""
    return list(self._tools.keys())

register 

register(tool_cls)

Register a tool class.

Raises:

Type Description
ValueError

If the class has no non-empty name attribute.
Source code in teff/tool/registry.py 
29
30
31
32
33
34
35
36
37
38
39
def register(self, tool_cls: type[Tool]) -> None:
    """Register a tool class.

    Raises:
        ValueError: If the class has no non-empty *name* attribute.
    """
    if not hasattr(tool_cls, "name") or not tool_cls.name:
        raise ValueError(
            f"Tool class {tool_cls.__name__} must have a non-empty 'name' attribute"
        )
    self._tools[tool_cls.name] = tool_cls

mcp_tools async 

mcp_tools(url=None, command=None, *, env=None, cwd=None, client_info=None)

Connect to an MCP server and yield its tools as Teff :class:Tool\s.

Exactly one of url or command must be given:

  • url: Streamable HTTP endpoint of an MCP server, e.g. http://localhost:8000/mcp.
  • command: Subprocess invocation for a stdio server, e.g. ["uvx", "mcp-server-git"] or ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"].

The session stays open for the duration of the async with block; tools keep working until it exits, after which the connection is closed.

Parameters:

Name Type Description Default
url str | None

Streamable HTTP endpoint.

 None
command list[str] | None

Stdio server command (list of argv tokens).

 None
env dict[str, str] | None

Optional extra environment variables for stdio servers.

 None
cwd str | None

Optional working directory for stdio servers.

 None
client_info dict | None

Optional dict overrides for the client Implementation advertised to the server.

 None

Yields:

Type Description
AsyncIterator[list[McpTool]]

A list of :class:McpTool instances, one per server tool.
Source code in teff/tool/mcp/bridge.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
@contextlib.asynccontextmanager
async def mcp_tools(
    url: str | None = None,
    command: list[str] | None = None,
    *,
    env: dict[str, str] | None = None,
    cwd: str | None = None,
    client_info: dict | None = None,
) -> AsyncIterator[list[McpTool]]:
    """Connect to an MCP server and yield its tools as Teff :class:`Tool`\\s.

    Exactly one of *url* or *command* must be given:

    - ``url``: Streamable HTTP endpoint of an MCP server, e.g.
      ``http://localhost:8000/mcp``.
    - ``command``: Subprocess invocation for a stdio server, e.g.
      ``["uvx", "mcp-server-git"]`` or
      ``["npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"]``.

    The session stays open for the duration of the ``async with`` block;
    tools keep working until it exits, after which the connection is closed.

    Args:
        url: Streamable HTTP endpoint.
        command: Stdio server command (list of argv tokens).
        env: Optional extra environment variables for stdio servers.
        cwd: Optional working directory for stdio servers.
        client_info: Optional dict overrides for the client
            ``Implementation`` advertised to the server.

    Yields:
        A list of :class:`McpTool` instances, one per server tool.
    """
    from mcp import StdioServerParameters
    from mcp.client.stdio import stdio_client
    from mcp.client.streamable_http import streamable_http_client

    if (url is None) == (command is None):
        raise ValueError("mcp_tools requires exactly one of 'url' or 'command'")

    stack = contextlib.AsyncExitStack()
    async with stack:
        if url is not None:
            # SDK stubs declare a wider tuple than runtime actually yields.
            read_stream, write_stream = await stack.enter_async_context(  # type: ignore[misc]
                streamable_http_client(url)
            )
        else:
            assert command is not None
            params = StdioServerParameters(
                command=command[0],
                args=command[1:],
                env=env,
                cwd=cwd,
            )
            read_stream, write_stream = await stack.enter_async_context(
                stdio_client(params)
            )

        session, tools = await _connect_tools(read_stream, write_stream, client_info)
        stack.push_async_callback(partial(session.__aexit__, None, None, None))
        yield tools

open_tools async 

open_tools(tools)

Expand :class:McpToolGroup entries in tools into their members.

Groups are opened on entry (so an async with open_tools(...) around a whole daemon loop keeps every server connected for its duration) and closed on exit. Plain tools pass through untouched.

Yields:

Type Description
AsyncIterator[list[Tool]]

A flat list of ready-to-call tools (group members replacing their
AsyncIterator[list[Tool]]

groups).
Source code in teff/tool/mcp/bridge.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@contextlib.asynccontextmanager
async def open_tools(
    tools: list[Tool] | list[McpToolGroup],
) -> AsyncIterator[list[Tool]]:
    """Expand :class:`McpToolGroup` entries in *tools* into their members.

    Groups are opened on entry (so an ``async with open_tools(...)`` around
    a whole daemon loop keeps every server connected for its duration) and
    closed on exit.  Plain tools pass through untouched.

    Yields:
        A flat list of ready-to-call tools (group members replacing their
        groups).
    """
    opened: list[McpToolGroup] = []
    ready: list[Tool] = []
    try:
        for tool in tools:
            if isinstance(tool, McpToolGroup):
                opened.append(tool)
                ready.extend(await tool.open())
            else:
                ready.append(tool)
        yield ready
    finally:
        for group in opened:
            await group.aclose()