MCP Lifecycle Management

Context manager and cleanup for MCP connections

MCP connections in PraisonAI Agents support context managers and explicit shutdown() so subprocesses, streams, and sockets close reliably.

MCP(...) accepts three equivalent construction forms — see Three equivalent forms.

from praisonaiagents import Agent, MCP

with MCP("uvx mcp-server-time") as mcp:
    agent = Agent(name="TimeAgent", instructions="Report the current time.", tools=mcp)
    agent.start("What time is it in UTC?")

The user runs an agent inside an MCP context manager; connections shut down cleanly when the session ends.

Quick Start

1. Simple Usage

   from praisonaiagents import Agent, MCP
   
   with MCP("uvx mcp-server-time") as mcp:
       agent = Agent(
           name="TimeAgent",
           instructions="Get the current time",
           tools=mcp
       )
       print(agent.start("What time is it?"))
   # Connection closed automatically

2. With Configuration

   import os
   from praisonaiagents import Agent, MCP
   
   with MCP(
       command="npx",
       args=["-y", "@modelcontextprotocol/server-brave-search"],
       env={"BRAVE_API_KEY": os.getenv("BRAVE_API_KEY")},
       timeout=30
   ) as mcp:
       agent = Agent(name="SearchAgent", tools=mcp)
       agent.start("Search for Python tutorials")

How It Works

Agent-Managed Cleanup

When you pass an MCP client through the constructor (tools=[MCP(...)]), agent.close() and agent.aclose() now walk the agent’s tools and shut down anything exposing .shutdown() / .aclose() — so the MCP subprocess and its background thread are cleaned up with the agent.

from praisonaiagents import Agent
from praisonaiagents.mcp import MCP

agent = Agent(
    name="Coder",
    tools=[MCP("uvx some-mcp-server")],
    instructions="Use MCP tools",
)
try:
    agent.start("do work")
finally:
    agent.close()   # now correctly shuts down MCP subprocesses/threads
                    # (async: `await agent.aclose()`)

Manual Cleanup

For cases where a context manager is not suitable:

from praisonaiagents import MCP

mcp = MCP("uvx mcp-server-time")

try:
    tools = mcp.get_tools()
finally:
    mcp.shutdown()

Request Cancellation

An MCP client can abort an in-flight request by sending a notifications/cancelled notification naming the requestId it wants to stop.

1. Send the request

   The client calls a tool with a JSON-RPC id it can reference later.

   { "jsonrpc": "2.0", "id": 42, "method": "tools/call", "params": { "name": "search" } }

2. Cancel it

   The client sends a cancellation notification with the same id as requestId.

   { "jsonrpc": "2.0", "method": "notifications/cancelled", "params": { "requestId": 42 } }

3. Receive the cancelled response

   The server cancels the running task and replies with a JSON-RPC error.

   { "jsonrpc": "2.0", "id": 42, "error": { "code": -32603, "message": "Request cancelled" } }

Lifecycle Methods

Authenticating the HTTP transport

When api_key is configured on the MCP HTTP-stream server, all of GET, POST, and DELETE require:

Authorization: Bearer <key>

Comparison uses constant-time hmac.compare_digest (timing-attack resistant). Missing or wrong tokens return 401 Unauthorized with {"error": "Unauthorized"}. DELETE returning 401 instead of 405 prevents information disclosure about whether sessions exist.

curl -H "Authorization: Bearer ${MCP_API_KEY:?Set MCP_API_KEY}" https://localhost:8080/mcp

__enter__ / __exit__

Context manager protocol for automatic resource management:

with MCP(command) as mcp:
    pass
# __exit__ calls shutdown() automatically

shutdown()

Explicitly close all connections and cleanup resources:

mcp = MCP("uvx mcp-server-time")
mcp.shutdown()

shutdown() closes an MCP connection. For stdio transports it enqueues a sentinel on the runner queue, joins the background thread, and lets the transport terminate its child process. For HTTP-stream / WebSocket transports it dispatches to close() (or aclose() if awaitable). It is idempotent — a second call is a no-op — and it never raises: cleanup errors are swallowed as best-effort.

Idempotency & retry

shutdown() is safe to call multiple times:

from praisonaiagents import MCP

mcp = MCP("uvx mcp-server-time")
mcp.shutdown()
mcp.shutdown()   # no-op — self.runner is None

If an in-flight tool call keeps the runner thread alive past the join timeout, the runner reference is kept so a later shutdown() can retry the join. The thread is a daemon, so a surviving runner never blocks interpreter exit.

__del__

Destructor ensures cleanup even if shutdown() was not called:

mcp = MCP("uvx mcp-server-time")
del mcp  # __del__ runs a non-blocking cleanup

Garbage-collection safety

A collected MCP never blocks the garbage collector.

Connection Types

MCP supports multiple connection types, all with proper cleanup:

# Stdio
with MCP("uvx mcp-server-time") as mcp: ...

# SSE
with MCP("http://localhost:8080/sse") as mcp: ...

# HTTP Stream — pass the full endpoint URL your server exposes (e.g. /mcp)
with MCP("http://localhost:8080/mcp") as mcp: ...

# WebSocket
with MCP("ws://localhost:8080") as mcp: ...

All four transports now close uniformly on shutdown(). HTTP-stream and WebSocket close via the close() fallback inside MCP.shutdown() — previously these two leaked subprocesses/threads on shutdown, while stdio and SSE cleaned up correctly.

Long-running processes

Gateways, HTTP invoke APIs, Jupyter kernels, and CI test runners create many Agent(mcp=...) instances over their lifetime. Every stdio MCP spawns a child process (npx / node) and a daemon runner thread. If those aren’t torn down when the agent is done, the host process leaks one child per MCP per agent.

The recommended pattern for long-lived hosts is a try/finally around every agent that carries MCP tools:

from praisonaiagents import Agent, MCP

def handle_request(prompt: str) -> str:
    agent = Agent(
        name="Assistant",
        tools=[MCP("uvx mcp-server-time")],
    )
    try:
        return agent.start(prompt)
    finally:
        agent.close()   # walks tools, calls MCP.shutdown() on each

Async equivalent:

import asyncio
from praisonaiagents import Agent, MCP

async def handle_request(prompt: str) -> str:
    async with Agent(
        name="Assistant",
        tools=[MCP("uvx mcp-server-time")],
    ) as agent:
        return await agent.achat(prompt)

Crash Recovery

A crashed stdio server respawns on the next call, up to 3 times, and the call that saw the crash is not replayed. See Auto-respawn on crash.

Best Practices

Related