Skip to content

kaboo_workflows.mcp

Model Context Protocol client/server lifecycle and transports.

kaboo_workflows.mcp

MCP server and client lifecycle management.

MCP_SERVER_TRANSPORT module-attribute 

MCP_SERVER_TRANSPORT = Literal['sse', 'streamable-http']

Transport types valid for :class:~kaboo_workflows.mcp.server.MCPServer.

stdio is excluded because it is a client-side transport where the client spawns the server as a subprocess and communicates over stdin/stdout pipes — there is no HTTP server to manage.

MCPLifecycle 

MCPLifecycle(server_ready_timeout=30)

Manages MCP server and client lifecycle ordering.

Initialize the MCPLifecycle.

Ensures servers are fully ready before clients connect, and clients are stopped before servers on shutdown.

Example::

lifecycle = MCPLifecycle()
lifecycle.add_server("postgres", pg_server)
lifecycle.add_client("pg_client", pg_client)

with lifecycle:
    # All servers started and ready, all clients connected
    agent = Agent(tools=[lifecycle.get_client("pg_client")])
    agent("Query the database")

# All cleaned up

Or without context manager::

lifecycle.start()
try:
    ...
finally:
    lifecycle.stop()

Parameters:

Name Type Description Default
server_ready_timeout float

Seconds to wait for each server to become ready.

 30
Source code in src/kaboo_workflows/mcp/lifecycle.py 
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
def __init__(self, server_ready_timeout: float = 30) -> None:
    """Initialize the MCPLifecycle.

    Ensures servers are fully ready before clients connect, and clients
    are stopped before servers on shutdown.

    Example::

        lifecycle = MCPLifecycle()
        lifecycle.add_server("postgres", pg_server)
        lifecycle.add_client("pg_client", pg_client)

        with lifecycle:
            # All servers started and ready, all clients connected
            agent = Agent(tools=[lifecycle.get_client("pg_client")])
            agent("Query the database")

        # All cleaned up

    Or without context manager::

        lifecycle.start()
        try:
            ...
        finally:
            lifecycle.stop()

    Args:
        server_ready_timeout: Seconds to wait for each server to become ready.
    """
    self._servers: dict[str, MCPServer] = {}
    self._clients: dict[str, StrandsMCPClient] = {}
    self._server_ready_timeout = server_ready_timeout
    self._started = False
    self._consumer_token = f"kaboo-mcp-lifecycle-{id(self)}"

servers property 

servers

Read-only view of registered servers.

clients property 

clients

Read-only view of registered clients.

add_server 

add_server(name, server)

Register an MCP server.

Parameters:

Name Type Description Default
name str

Unique server identifier.

 required
server MCPServer

The MCP server instance.

 required

Raises:

Type Description
ValueError

If a server with this name is already registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
def add_server(self, name: str, server: MCPServer) -> None:
    """Register an MCP server.

    Args:
        name: Unique server identifier.
        server: The MCP server instance.

    Raises:
        ValueError: If a server with this name is already registered.
    """
    if name in self._servers:
        raise ValueError(f"MCP server '{name}' is already registered")
    self._servers[name] = server

add_client 

add_client(name, client)

Register an MCP client.

Parameters:

Name Type Description Default
name str

Unique client identifier.

 required
client MCPClient

The strands MCP client instance.

 required

Raises:

Type Description
ValueError

If a client with this name is already registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
def add_client(self, name: str, client: StrandsMCPClient) -> None:
    """Register an MCP client.

    Args:
        name: Unique client identifier.
        client: The strands MCP client instance.

    Raises:
        ValueError: If a client with this name is already registered.
    """
    if name in self._clients:
        raise ValueError(f"MCP client '{name}' is already registered")
    self._clients[name] = client

get_server 

get_server(name)

Get a registered server by name.

Parameters:

Name Type Description Default
name str

Server identifier.

 required

Returns:

Type Description
MCPServer

The registered MCP server.

Raises:

Type Description
KeyError

If no server with this name is registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_server(self, name: str) -> MCPServer:
    """Get a registered server by name.

    Args:
        name: Server identifier.

    Returns:
        The registered MCP server.

    Raises:
        KeyError: If no server with this name is registered.
    """
    if name not in self._servers:
        raise KeyError(f"MCP server '{name}' not registered.\nAvailable: {list(self._servers)}")
    return self._servers[name]

get_client 

get_client(name)

Get a registered client by name.

Parameters:

Name Type Description Default
name str

Client identifier.

 required

Returns:

Type Description
MCPClient

The registered strands MCP client.

Raises:

Type Description
KeyError

If no client with this name is registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_client(self, name: str) -> StrandsMCPClient:
    """Get a registered client by name.

    Args:
        name: Client identifier.

    Returns:
        The registered strands MCP client.

    Raises:
        KeyError: If no client with this name is registered.
    """
    if name not in self._clients:
        raise KeyError(f"MCP client '{name}' not registered.\nAvailable: {list(self._clients)}")
    return self._clients[name]

start 

start()

Start all servers and wait for readiness.

Idempotent: if already started, returns immediately. load() calls this before creating agents (so MCP clients can connect), and the context manager calls it again on enter — the second call is a no-op. The context manager is still needed for graceful shutdown via stop().

Clients are not started here — strands automatically starts MCPClient instances when they are registered as tool providers on an Agent. Starting them here would cause a "session is currently running" error when the Agent tries to start them again. Instead we pin each client with a permanent lifecycle consumer so its session is not torn down when consuming Agents are garbage-collected.

Raises:

Type Description
RuntimeError

If any server fails to start or become ready.
Source code in src/kaboo_workflows/mcp/lifecycle.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
def start(self) -> None:
    """Start all servers and wait for readiness.

    **Idempotent**: if already started, returns immediately.
    ``load()`` calls this before creating agents (so MCP clients can
    connect), and the context manager calls it again on enter — the
    second call is a no-op.  The context manager is still needed for
    **graceful shutdown** via ``stop()``.

    Clients are **not** started here — strands automatically starts
    MCPClient instances when they are registered as tool providers
    on an Agent. Starting them here would cause a "session is currently
    running" error when the Agent tries to start them again. Instead we
    pin each client with a permanent lifecycle consumer so its session
    is not torn down when consuming Agents are garbage-collected.

    Raises:
        RuntimeError: If any server fails to start or become ready.
    """
    if self._started:
        return

    # Phase 1: Start all servers
    for name, server in self._servers.items():
        logger.info("server=<%s> | starting MCP server", name)
        server.start()

    # Phase 2: Wait for all servers to be ready
    for name, server in self._servers.items():
        if not server.wait_ready(timeout=self._server_ready_timeout):
            raise RuntimeError(
                f"MCP server '{name}' did not become ready within {self._server_ready_timeout}s"
            )
        logger.info("server=<%s> | MCP server is ready", name)

    # Phase 3: pin every client with a permanent consumer so its session
    # survives agent garbage collection (esp. across an interrupt/resume
    # boundary). Strands reference-counts clients per consuming Agent and
    # tears the session down when the count hits zero; stop() clears it.
    for name, client in self._clients.items():
        add_consumer = getattr(client, "add_consumer", None)
        if callable(add_consumer):
            add_consumer(self._consumer_token)
            logger.debug("client=<%s> | pinned lifecycle consumer", name)

    self._started = True

stop 

stop()

Stop all clients first, then all servers.

Clients that were never started (e.g., never registered on an Agent) are skipped gracefully.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
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
def stop(self) -> None:
    """Stop all clients first, then all servers.

    Clients that were never started (e.g., never registered on an Agent)
    are skipped gracefully.
    """
    if not self._started:
        return

    # Phase 1: Stop all clients
    for name, client in self._clients.items():
        try:
            # Normal shutdown — no exception context (matches __exit__ protocol)
            client.stop(exc_type=None, exc_val=None, exc_tb=None)
            logger.info("client=<%s> | MCP client stopped", name)
        except Exception:
            logger.warning("client=<%s> | failed to stop MCP client", name, exc_info=True)

    # Phase 2: Stop all servers
    for name, server in self._servers.items():
        try:
            server.stop()
            logger.info("server=<%s> | MCP server stopped", name)
        except Exception:
            logger.warning("server=<%s> | failed to stop MCP server", name, exc_info=True)

    self._started = False

__enter__ 

__enter__()

Start lifecycle on context entry.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
202
203
204
205
def __enter__(self) -> MCPLifecycle:
    """Start lifecycle on context entry."""
    self.start()
    return self

__exit__ 

__exit__(exc_type, exc_val, exc_tb)

Stop lifecycle on context exit.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
207
208
209
210
211
212
213
214
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Stop lifecycle on context exit."""
    self.stop()

__aenter__ async 

__aenter__()

Async context entry — delegates to sync :meth:start.

Useful with Starlette / ASGI lifespan::

@asynccontextmanager
async def lifespan(app):
    async with lifecycle:
        yield
Source code in src/kaboo_workflows/mcp/lifecycle.py 
216
217
218
219
220
221
222
223
224
225
226
227
async def __aenter__(self) -> MCPLifecycle:
    """Async context entry — delegates to sync :meth:`start`.

    Useful with Starlette / ASGI lifespan::

        @asynccontextmanager
        async def lifespan(app):
            async with lifecycle:
                yield
    """
    self.start()
    return self

__aexit__ async 

__aexit__(exc_type, exc_val, exc_tb)

Async context exit — delegates to sync :meth:stop.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
229
230
231
232
233
234
235
236
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context exit — delegates to sync :meth:`stop`."""
    self.stop()

MCPServer 

MCPServer(
    *,
    name,
    host="127.0.0.1",
    port=8000,
    transport=DEFAULT_TRANSPORT,
    server_params=None,
)

Bases: ABC

Abstract base for kaboo_workflows MCP servers.

Initialize the MCPServer.

Subclasses implement _register_tools() to register tools on the FastMCP instance. The base class manages background-thread lifecycle and readiness signaling.

Parameters:

Name Type Description Default
name str

Unique server identifier.

 required
host str

Bind address for the HTTP transport.

 '127.0.0.1'
port int

Bind port for the HTTP transport.

 8000
transport MCP_SERVER_TRANSPORT

MCP server transport type (streamable-http or sse).

 DEFAULT_TRANSPORT
server_params dict[str, Any] | None

Extra keyword arguments forwarded to FastMCP().

 None
Source code in src/kaboo_workflows/mcp/server.py 
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
def __init__(
    self,
    *,
    name: str,
    host: str = "127.0.0.1",
    port: int = 8000,
    transport: MCP_SERVER_TRANSPORT = DEFAULT_TRANSPORT,
    server_params: dict[str, Any] | None = None,
) -> None:
    """Initialize the MCPServer.

    Subclasses implement ``_register_tools()`` to register tools on the
    ``FastMCP`` instance.  The base class manages background-thread
    lifecycle and readiness signaling.

    Args:
        name: Unique server identifier.
        host: Bind address for the HTTP transport.
        port: Bind port for the HTTP transport.
        transport: MCP server transport type (``streamable-http`` or ``sse``).
        server_params: Extra keyword arguments forwarded to ``FastMCP()``.
    """
    self.name = name
    self.host = host
    self.port = port
    self.transport = transport
    self.server_params = server_params or {}
    self._mcp: FastMCP | None = None
    self._thread: threading.Thread | None = None
    self._ready = threading.Event()
    self._error: BaseException | None = None
    self._uvicorn_server: uvicorn.Server | None = None

url property 

url

Base URL of this server (for client transport).

is_running property 

is_running

True if the server thread is alive.

create_server 

create_server()

Build the FastMCP instance and register tools.

The result is cached — calling twice returns the same instance.

Source code in src/kaboo_workflows/mcp/server.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def create_server(self) -> FastMCP:
    """Build the ``FastMCP`` instance and register tools.

    The result is cached — calling twice returns the same instance.
    """
    if self._mcp is not None:
        return self._mcp

    from mcp.server.fastmcp import FastMCP as _FastMCP

    mcp = _FastMCP(
        self.name,
        host=self.host,
        port=self.port,
        stateless_http=True,
        json_response=True,
        log_level="WARNING",
        **self.server_params,
    )
    self._register_tools(mcp)
    self._mcp = mcp
    return mcp

run 

run()

Start the server blocking (for standalone CLI usage).

In the main thread FastMCP.run() installs signal handlers so that Ctrl-C triggers a graceful uvicorn shutdown.

Source code in src/kaboo_workflows/mcp/server.py 
157
158
159
160
161
162
163
164
def run(self) -> None:
    """Start the server blocking (for standalone CLI usage).

    In the main thread ``FastMCP.run()`` installs signal handlers so
    that Ctrl-C triggers a graceful uvicorn shutdown.
    """
    mcp = self.create_server()
    mcp.run(transport=self.transport)

start 

start()

Start the server in a background daemon thread.

Creates its own uvicorn.Server instead of delegating to FastMCP.run(). This keeps a reference to the server so that :meth:stop can trigger a graceful shutdown via uvicorn.Server.should_exit.

Source code in src/kaboo_workflows/mcp/server.py 
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
def start(self) -> None:
    """Start the server in a background daemon thread.

    Creates its own ``uvicorn.Server`` instead of delegating to
    ``FastMCP.run()``.  This keeps a reference to the server so that
    :meth:`stop` can trigger a graceful shutdown via
    ``uvicorn.Server.should_exit``.
    """
    if self.is_running:
        return
    self._ready.clear()
    self._error = None

    mcp = self.create_server()
    asgi_app = self._get_asgi_app(mcp)

    import uvicorn as _uvicorn

    config = _uvicorn.Config(
        asgi_app,
        host=self.host,
        port=self.port,
        log_level="warning",
    )
    self._uvicorn_server = _uvicorn.Server(config)

    def _target() -> None:
        try:
            asyncio.run(self._uvicorn_server.serve())  # ty: ignore
        except BaseException as exc:
            self._error = exc
            self._ready.set()

    self._thread = threading.Thread(
        target=_target,
        name=f"mcp-{self.name}",
        daemon=True,
    )
    self._thread.start()

wait_ready 

wait_ready(timeout=30)

Wait for the server to be ready by polling the TCP port.

Returns:

Type Description
bool

True if server is ready, False if timed out.

Raises:

Type Description
RuntimeError

If the server thread died before becoming ready.
Source code in src/kaboo_workflows/mcp/server.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def wait_ready(self, timeout: float = 30) -> bool:
    """Wait for the server to be ready by polling the TCP port.

    Returns:
        True if server is ready, False if timed out.

    Raises:
        RuntimeError: If the server thread died before becoming ready.
    """
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if self._error is not None:
            raise RuntimeError(
                f"MCP server '{self.name}' failed to start: {self._error}"
            ) from self._error
        if self._thread is not None and not self._thread.is_alive():
            raise RuntimeError(f"MCP server '{self.name}' thread exited unexpectedly")
        try:
            with socket.create_connection((self.host, self.port), timeout=1):
                self._ready.set()
                return True
        except OSError:
            time.sleep(0.1)
    return False

stop 

stop()

Stop the server and clean up the background thread.

Signals uvicorn.Server.should_exit which triggers a graceful drain (stop accepting new connections, finish in-flight requests). If the thread does not exit within :attr:STOP_TIMEOUT seconds, force_exit is set to skip connection draining. After a further :attr:STOP_FORCE_TIMEOUT seconds the thread is abandoned as a daemon thread and will be reaped when the process exits.

Source code in src/kaboo_workflows/mcp/server.py 
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
def stop(self) -> None:
    """Stop the server and clean up the background thread.

    Signals ``uvicorn.Server.should_exit`` which triggers a graceful
    drain (stop accepting new connections, finish in-flight requests).
    If the thread does not exit within :attr:`STOP_TIMEOUT` seconds,
    ``force_exit`` is set to skip connection draining.  After a
    further :attr:`STOP_FORCE_TIMEOUT` seconds the thread is
    abandoned as a daemon thread and will be reaped when the process
    exits.
    """
    if self._thread is not None and self._thread.is_alive():
        if self._uvicorn_server is not None:
            # Graceful phase: ask uvicorn to stop accepting and drain.
            self._uvicorn_server.should_exit = True
            self._thread.join(timeout=self.STOP_TIMEOUT)

            if self._thread.is_alive():
                # Forceful phase: skip connection draining.
                logger.info(
                    "server=<%s>, timeout=<%s> | forcing exit after graceful stop timeout",
                    self.name,
                    self.STOP_TIMEOUT,
                )
                self._uvicorn_server.force_exit = True
                self._thread.join(timeout=self.STOP_FORCE_TIMEOUT)

        if self._thread.is_alive():
            logger.warning(
                "server=<%s> | thread did not stop, daemon will be reaped at exit", self.name
            )

    self._uvicorn_server = None
    self._mcp = None
    self._thread = None
    self._ready.clear()

create_mcp_client 

create_mcp_client(
    *,
    server=None,
    url=None,
    command=None,
    transport=DEFAULT_TRANSPORT,
    transport_options=None,
    **kwargs,
)

Create a strands MCPClient from connection configuration.

Exactly one of server, url, or command must be provided.

Parameters:

Name Type Description Default
server MCPServer | None

A managed MCPServer instance (connects via its URL).

 None
url str | None

External MCP server URL (for SSE or streamable-http).

 None
command list[str] | None

Command to start an MCP server subprocess (stdio transport).

 None
transport MCP_TRANSPORT

Override transport type ("stdio", "sse", "streamable-http"). Auto-detected if not specified.

 DEFAULT_TRANSPORT
transport_options dict[str, Any] | None

Extra kwargs forwarded to the transport factory. These are transport-specific — see each transport function for available options:

stdio: env, cwd, encoding, encoding_error_handler

sse: headers, timeout, sse_read_timeout, auth, httpx_client_factory

streamable-http: headers, http_client (pre-configured httpx.AsyncClient), terminate_on_close

 None
**kwargs Any

Additional kwargs forwarded to strands MCPClient (startup_timeout, tool_filters, prefix, elicitation_callback, tasks_config, etc.).

 {}

Returns:

Type Description
MCPClient

A strands MCPClient instance.

Raises:

Type Description
ValueError

If connection parameters are ambiguous.
Source code in src/kaboo_workflows/mcp/client.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
def create_mcp_client(
    *,
    server: MCPServer | None = None,
    url: str | None = None,
    command: list[str] | None = None,
    transport: MCP_TRANSPORT = DEFAULT_TRANSPORT,
    transport_options: dict[str, Any] | None = None,
    **kwargs: Any,
) -> MCPClient:
    """Create a strands MCPClient from connection configuration.

    Exactly one of server, url, or command must be provided.

    Args:
        server: A managed MCPServer instance (connects via its URL).
        url: External MCP server URL (for SSE or streamable-http).
        command: Command to start an MCP server subprocess (stdio transport).
        transport: Override transport type ("stdio", "sse", "streamable-http").
            Auto-detected if not specified.
        transport_options: Extra kwargs forwarded to the transport factory.
            These are transport-specific — see each transport function for
            available options:

            **stdio**: ``env``, ``cwd``, ``encoding``, ``encoding_error_handler``

            **sse**: ``headers``, ``timeout``, ``sse_read_timeout``, ``auth``,
            ``httpx_client_factory``

            **streamable-http**: ``headers``, ``http_client`` (pre-configured
            ``httpx.AsyncClient``), ``terminate_on_close``

        **kwargs: Additional kwargs forwarded to strands MCPClient
            (startup_timeout, tool_filters, prefix, elicitation_callback,
            tasks_config, etc.).

    Returns:
        A strands MCPClient instance.

    Raises:
        ValueError: If connection parameters are ambiguous.
    """
    modes = sum(x is not None for x in [server, url, command])
    if modes != 1:
        raise ValueError(
            f"Exactly one of server, url, or command must be provided (got {modes}).\n"
            "server=MCPServer for managed servers, url=str for external HTTP, "
            "command=list[str] for subprocess stdio."
        )

    opts = transport_options or {}

    if server is not None:
        transport_callable = _transport_for_http(server.url, transport, opts, allow_stdio=False)
    elif url is not None:
        transport_callable = _transport_for_http(url, transport, opts, allow_stdio=True)
    else:
        # command is guaranteed non-None by the modes == 1 check above.
        transport_callable = stdio_transport(command, **opts)  # ty: ignore

    return _make_strands_client(transport_callable=transport_callable, **kwargs)

create_mcp_server 

create_mcp_server(
    *,
    name,
    tools,
    host="127.0.0.1",
    port=8000,
    transport=DEFAULT_TRANSPORT,
    server_params=None,
)

Create an MCP server from a list of callables — no subclassing needed.

Each callable (sync or async) is registered as a tool on the underlying FastMCP instance. For advanced use (custom state, routes, resources), subclass :class:MCPServer directly.

Example::

def get_weather(city: str) -> str:
    return f"Sunny in {city}"


async def query_db(sql: str) -> str: ...


server = create_mcp_server(name="weather", tools=[get_weather, query_db], port=8001)
server.start()

Parameters:

Name Type Description Default
name str

Unique server identifier.

 required
tools list[Callable[..., Any]]

Callables to register as MCP tools.

 required
host str

Bind address (default 127.0.0.1).

 '127.0.0.1'
port int

Bind port (default 8000).

 8000
transport MCP_SERVER_TRANSPORT

Server transport type (streamable-http or sse).

 DEFAULT_TRANSPORT
server_params dict[str, Any] | None

Extra kwargs forwarded to FastMCP().

 None

Returns:

Type Description
MCPServer

A ready-to-use :class:MCPServer instance.
Source code in src/kaboo_workflows/mcp/server.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def create_mcp_server(
    *,
    name: str,
    tools: list[Callable[..., Any]],
    host: str = "127.0.0.1",
    port: int = 8000,
    transport: MCP_SERVER_TRANSPORT = DEFAULT_TRANSPORT,
    server_params: dict[str, Any] | None = None,
) -> MCPServer:
    """Create an MCP server from a list of callables — no subclassing needed.

    Each callable (sync or async) is registered as a tool on the underlying
    ``FastMCP`` instance.  For advanced use (custom state, routes, resources),
    subclass :class:`MCPServer` directly.

    Example::

        def get_weather(city: str) -> str:
            return f"Sunny in {city}"


        async def query_db(sql: str) -> str: ...


        server = create_mcp_server(name="weather", tools=[get_weather, query_db], port=8001)
        server.start()

    Args:
        name: Unique server identifier.
        tools: Callables to register as MCP tools.
        host: Bind address (default ``127.0.0.1``).
        port: Bind port (default ``8000``).
        transport: Server transport type (``streamable-http`` or ``sse``).
        server_params: Extra kwargs forwarded to ``FastMCP()``.

    Returns:
        A ready-to-use :class:`MCPServer` instance.
    """
    tool_fns = list(tools)

    class _FactoryServer(MCPServer):
        def _register_tools(self, mcp: FastMCP) -> None:
            for fn in tool_fns:
                mcp.tool()(fn)

    return _FactoryServer(
        name=name,
        host=host,
        port=port,
        transport=transport,
        server_params=server_params,
    )

sse_transport 

sse_transport(
    url,
    headers=None,
    *,
    timeout=5,
    sse_read_timeout=300,
    auth=None,
    httpx_client_factory=None,
)

Create an SSE (Server-Sent Events) transport callable.

Parameters:

Name Type Description Default
url str

SSE endpoint URL.

 required
headers dict[str, Any] | None

Optional HTTP headers.

 None
timeout float

HTTP timeout in seconds (default: 5).

 5
sse_read_timeout float

Timeout waiting for SSE events in seconds (default: 300).

 300
auth Any | None

Optional httpx.Auth instance (e.g., OAuth provider).

 None
httpx_client_factory Any | None

Optional factory for creating httpx client.

 None

Returns:

Type Description
Callable[[], Any]

Transport callable for strands MCPClient.

Raises:

Type Description
ValueError

If url is empty.
Source code in src/kaboo_workflows/mcp/transports.py 
 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
def sse_transport(
    url: str,
    headers: dict[str, Any] | None = None,
    *,
    timeout: float = 5,
    sse_read_timeout: float = 300,
    auth: Any | None = None,
    httpx_client_factory: Any | None = None,
) -> Callable[[], Any]:
    """Create an SSE (Server-Sent Events) transport callable.

    Args:
        url: SSE endpoint URL.
        headers: Optional HTTP headers.
        timeout: HTTP timeout in seconds (default: 5).
        sse_read_timeout: Timeout waiting for SSE events in seconds (default: 300).
        auth: Optional httpx.Auth instance (e.g., OAuth provider).
        httpx_client_factory: Optional factory for creating httpx client.

    Returns:
        Transport callable for strands MCPClient.

    Raises:
        ValueError: If url is empty.
    """
    if not url:
        raise ValueError("url must be a non-empty string")

    captured_headers = headers or {}
    captured_timeout = timeout
    captured_sse_read_timeout = sse_read_timeout
    captured_auth = auth
    captured_httpx_client_factory = httpx_client_factory

    def factory() -> Any:
        from mcp.client.sse import sse_client

        kwargs: dict[str, Any] = {
            "url": url,
            "headers": captured_headers,
            "timeout": captured_timeout,
            "sse_read_timeout": captured_sse_read_timeout,
        }
        if captured_auth is not None:
            kwargs["auth"] = captured_auth
        if captured_httpx_client_factory is not None:
            kwargs["httpx_client_factory"] = captured_httpx_client_factory
        return sse_client(**kwargs)

    return factory

stdio_transport 

stdio_transport(
    command,
    env=None,
    *,
    cwd=None,
    encoding="utf-8",
    encoding_error_handler="strict",
)

Create a stdio transport callable for a subprocess MCP server.

Parameters:

Name Type Description Default
command list[str]

Command to start the MCP server (e.g., ["python", "-m", "myserver"]).

 required
env dict[str, str] | None

Optional environment variables for the subprocess.

 None
cwd str | Path | None

Working directory for the subprocess.

 None
encoding str

Text encoding for messages (default: utf-8).

 'utf-8'
encoding_error_handler Literal['strict', 'ignore', 'replace']

How to handle encoding errors (default: strict).

 'strict'

Returns:

Type Description
Callable[[], Any]

Transport callable for strands MCPClient.

Raises:

Type Description
ValueError

If command is empty.
Source code in src/kaboo_workflows/mcp/transports.py 
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
def stdio_transport(
    command: list[str],
    env: dict[str, str] | None = None,
    *,
    cwd: str | Path | None = None,
    encoding: str = "utf-8",
    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict",
) -> Callable[[], Any]:
    """Create a stdio transport callable for a subprocess MCP server.

    Args:
        command: Command to start the MCP server (e.g., ["python", "-m", "myserver"]).
        env: Optional environment variables for the subprocess.
        cwd: Working directory for the subprocess.
        encoding: Text encoding for messages (default: utf-8).
        encoding_error_handler: How to handle encoding errors (default: strict).

    Returns:
        Transport callable for strands MCPClient.

    Raises:
        ValueError: If command is empty.
    """
    if not command:
        raise ValueError("command must be a non-empty list (e.g., ['python', '-m', 'myserver'])")

    captured_command = list(command)
    captured_env = dict(env) if env is not None else None
    captured_cwd = cwd
    captured_encoding = encoding
    captured_encoding_error_handler = encoding_error_handler

    def factory() -> Any:
        from mcp.client.stdio import StdioServerParameters, stdio_client

        params = StdioServerParameters(
            command=captured_command[0],
            args=captured_command[1:],
            env=captured_env,
            cwd=captured_cwd,
            encoding=captured_encoding,
            encoding_error_handler=captured_encoding_error_handler,
        )
        return stdio_client(params)

    return factory

streamable_http_transport 

streamable_http_transport(
    url,
    headers=None,
    *,
    http_client=None,
    terminate_on_close=True,
)

Create a streamable HTTP transport callable.

For full control (auth, timeouts, custom TLS, etc.), pass a pre-configured httpx.AsyncClient via http_client. When http_client is provided, headers is ignored (configure headers on the client directly).

Parameters:

Name Type Description Default
url str

HTTP endpoint URL (e.g., "http://localhost:8000/mcp").

 required
headers dict[str, str] | None

Optional HTTP headers. Ignored when http_client is provided.

 None
http_client Any | None

Optional pre-configured httpx.AsyncClient.

 None
terminate_on_close bool

Send DELETE to close session (default: True).

 True

Returns:

Type Description
Callable[[], Any]

Transport callable for strands MCPClient.

Raises:

Type Description
ValueError

If url is empty.
Source code in src/kaboo_workflows/mcp/transports.py 
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
def streamable_http_transport(
    url: str,
    headers: dict[str, str] | None = None,
    *,
    http_client: Any | None = None,
    terminate_on_close: bool = True,
) -> Callable[[], Any]:
    """Create a streamable HTTP transport callable.

    For full control (auth, timeouts, custom TLS, etc.), pass a pre-configured
    ``httpx.AsyncClient`` via ``http_client``. When ``http_client`` is provided,
    ``headers`` is ignored (configure headers on the client directly).

    Args:
        url: HTTP endpoint URL (e.g., "http://localhost:8000/mcp").
        headers: Optional HTTP headers. Ignored when ``http_client`` is provided.
        http_client: Optional pre-configured ``httpx.AsyncClient``.
        terminate_on_close: Send DELETE to close session (default: True).

    Returns:
        Transport callable for strands MCPClient.

    Raises:
        ValueError: If url is empty.
    """
    if not url:
        raise ValueError("url must be a non-empty string")

    captured_headers = dict(headers) if headers else None
    captured_http_client = http_client
    captured_terminate_on_close = terminate_on_close

    def factory() -> Any:
        from mcp.client.streamable_http import streamable_http_client

        if captured_http_client is not None:
            return streamable_http_client(
                url=url,
                http_client=captured_http_client,
                terminate_on_close=captured_terminate_on_close,
            )
        if captured_headers:
            import httpx

            client = httpx.AsyncClient(headers=captured_headers)
            return streamable_http_client(
                url=url,
                http_client=client,
                terminate_on_close=captured_terminate_on_close,
            )
        return streamable_http_client(url=url, terminate_on_close=captured_terminate_on_close)

    return factory