teff.tool.mcp¶

teff.tool.mcp ¶

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

Public surface of the MCP integration:

:class:~teff.tool.mcp.bridge.McpToolGroup — a lazily-opened connection to one MCP server whose tools become <id>__<tool> members.
:func:~teff.tool.mcp.bridge.mcp_tools / :func:~teff.tool.mcp.bridge.open_tools — connection helpers.
:class:~teff.tool.mcp.presets.McpPreset and :data:MCP_PRESETS — ready-made launch configs for known servers; subclass McpPreset and it is registered automatically (each subclass needs a unique name).

Modules:

Name	Description
`bridge`	Bridge to Model Context Protocol (MCP) servers.
`presets`	Ready-made launch configs for known MCP servers.

Classes:

Name	Description
`FetchPreset`	Fetch: download web content as Markdown.
`GitPreset`	Git: read and edit Git repositories.
`GmailPreset`	Gmail: search, read, draft and send messages.
`GoogleCalendarPreset`	Google Calendar: read, create and update events.
`GoogleDrivePreset`	Google Drive: file search, read, create, upload.
`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.
`SQLitePreset`	SQLite: inspect and query SQLite databases.
`TimePreset`	Time / Timezone: current time and timezone conversion.

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.

FetchPreset ¶

Bases: McpPreset

Fetch: download web content as Markdown.

Source code in teff/tool/mcp/presets.py

class FetchPreset(McpPreset):
    """Fetch: download web content as Markdown."""

    name = "fetch"
    command = ["uvx", "mcp-server-fetch"]
    description = "Download web content as Markdown (via uvx)."

GitPreset ¶

Bases: McpPreset

Git: read and edit Git repositories.

Source code in teff/tool/mcp/presets.py

class GitPreset(McpPreset):
    """Git: read and edit Git repositories."""

    name = "git"
    command = ["uvx", "mcp-server-git"]
    description = "Read and edit Git repositories (via uvx)."

GmailPreset ¶

Bases: McpPreset

Gmail: search, read, draft and send messages.

Source code in teff/tool/mcp/presets.py

class GmailPreset(McpPreset):
    """Gmail: search, read, draft and send messages."""

    name = "gmail"
    command = ["npx", "-y", "@google/mcp-server-gmail"]
    env = {
        "GOOGLE_API_KEY": "",
        "GOOGLE_GMAIL_CLIENT_ID": "",
        "GOOGLE_GMAIL_CLIENT_SECRET": "",
        "GOOGLE_GMAIL_REFRESH_TOKEN": "",
    }
    description = "Gmail search, read, draft and send messages."

GoogleCalendarPreset ¶

Bases: McpPreset

Google Calendar: read, create and update events.

Source code in teff/tool/mcp/presets.py

class GoogleCalendarPreset(McpPreset):
    """Google Calendar: read, create and update events."""

    name = "google_calendar"
    command = ["npx", "-y", "@google/mcp-server-google-calendar"]
    env = {
        "GOOGLE_API_KEY": "",
        "GOOGLE_CALENDAR_CLIENT_ID": "",
        "GOOGLE_CALENDAR_CLIENT_SECRET": "",
        "GOOGLE_CALENDAR_REFRESH_TOKEN": "",
    }
    description = "Google Calendar read, create and update events."

GoogleDrivePreset ¶

Bases: McpPreset

Google Drive: file search, read, create, upload.

Source code in teff/tool/mcp/presets.py

class GoogleDrivePreset(McpPreset):
    """Google Drive: file search, read, create, upload."""

    name = "google_drive"
    command = ["npx", "-y", "@google/mcp-server-google-drive"]
    env = {
        "GOOGLE_API_KEY": "",
        "GOOGLE_DRIVE_CLIENT_ID": "",
        "GOOGLE_DRIVE_CLIENT_SECRET": "",
        "GOOGLE_DRIVE_REFRESH_TOKEN": "",
    }
    description = "Google Drive file search, read, create, upload."

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

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

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

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

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

@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

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

SQLitePreset ¶

Bases: McpPreset

SQLite: inspect and query SQLite databases.

Source code in teff/tool/mcp/presets.py

class SQLitePreset(McpPreset):
    """SQLite: inspect and query SQLite databases."""

    name = "sqlite"
    command = ["uvx", "mcp-server-sqlite"]
    description = "Inspect and query SQLite databases (via uvx)."

TimePreset ¶

Bases: McpPreset

Time / Timezone: current time and timezone conversion.

Source code in teff/tool/mcp/presets.py

class TimePreset(McpPreset):
    """Time / Timezone: current time and timezone conversion."""

    name = "time"
    command = ["uvx", "mcp-server-time"]
    description = "Current time and timezone conversion (via uvx)."

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

@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

@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()

teff.tool.mcp¶

teff.tool.mcp ¶

FetchPreset ¶

GitPreset ¶

GmailPreset ¶

GoogleCalendarPreset ¶

GoogleDrivePreset ¶

McpPreset ¶

McpTool ¶

McpToolGroup ¶

aclose async ¶

from_preset classmethod ¶

open async ¶

SQLitePreset ¶

TimePreset ¶

mcp_tools async ¶

open_tools async ¶

aclose `async` ¶

from_preset `classmethod` ¶

open `async` ¶

mcp_tools `async` ¶

open_tools `async` ¶