Implementation:Microsoft Agent framework Agent Async Context Manager

Knowledge Sources	Microsoft Agent Framework Python contextlib
Domains	Agent_Architecture
Last Updated	2026-02-11 17:00 GMT

Overview

Implementation of the async context manager protocol (__aenter__ / __aexit__) on the Agent class, enabling the async with Agent(...) as agent: pattern for automatic resource lifecycle management of MCP tools and chat client connections.

Description

The Agent class implements __aenter__ and __aexit__ to manage the lifecycle of resources that require explicit cleanup. During entry, the agent inspects its client and all registered mcp_tools and enters any that are AbstractAsyncContextManager instances into an internal AsyncExitStack. During exit, the stack is closed, triggering orderly teardown of all registered context managers in LIFO order.

Usage

Use the async with pattern whenever the agent holds MCP tools or a client that implements the async context manager protocol. This is the recommended way to instantiate and run agents to prevent resource leaks.

Code Reference

Source Location

• Repository: agent-framework
• File: python/packages/core/agent_framework/_agents.py
• Lines: L730-762

Signature

async def __aenter__(self) -> Self:
    """Enter the async context manager.

    If any of the client or local_mcp_tools are context managers,
    they will be entered into the async exit stack to ensure proper cleanup.

    Returns:
        The Agent instance.
    """
    for context_manager in chain([self.client], self.mcp_tools):
        if isinstance(context_manager, AbstractAsyncContextManager):
            await self._async_exit_stack.enter_async_context(context_manager)
    return self

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None:
    """Exit the async context manager.

    Close the async exit stack to ensure all context managers are exited properly.
    """
    await self._async_exit_stack.aclose()

Import

from agent_framework import Agent

I/O Contract

Inputs

Method	Name	Type	Required	Description
__aenter__	self	Agent	Yes	The Agent instance. No additional parameters.
__aexit__	exc_type	None	Yes	The exception type if an exception was raised, None otherwise.
__aexit__	exc_val	None	Yes	The exception value if an exception was raised, None otherwise.
__aexit__	exc_tb	Any	Yes	The exception traceback if an exception was raised, None otherwise.

Outputs

Method	Type	Description
__aenter__	Self	Returns the same Agent instance, enabling the as agent binding in the async with statement.
__aexit__	None	Returns None. Does not suppress exceptions.

Internal State

Attribute	Type	Description
_async_exit_stack	AsyncExitStack	Initialized in __init__. Accumulates context managers during __aenter__ and closes them all during __aexit__.
client	SupportsChatGetResponse	The chat client. Entered into the exit stack if it is an AbstractAsyncContextManager.
mcp_tools	list[MCPTool]	MCP tool instances. Each is entered into the exit stack if it is an AbstractAsyncContextManager.

Usage Examples

Basic Agent Lifecycle

from agent_framework import Agent
from agent_framework.openai import OpenAIResponsesClient

client = OpenAIResponsesClient(model_id="gpt-4o")

async with Agent(client=client, instructions="You are a helpful assistant.", tools=[tool1]) as agent:
    response = await agent.run("Hello")
    print(response.text)
# Resources automatically cleaned up

With MCP Tools

from agent_framework import Agent
from agent_framework.openai import OpenAIResponsesClient
from agent_framework import MCPTool

client = OpenAIResponsesClient(model_id="gpt-4o")
mcp_tool = MCPTool(command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"])

async with Agent(client=client, instructions="You are a file assistant.", tools=[mcp_tool]) as agent:
    response = await agent.run("List the files in /tmp")
    print(response.text)
# MCP server subprocess and client connection are cleaned up

Without Context Manager (Not Recommended)

from agent_framework import Agent
from agent_framework.openai import OpenAIResponsesClient

client = OpenAIResponsesClient(model_id="gpt-4o")
agent = Agent(client=client, instructions="You are a helpful assistant.")

# Must manually enter and exit the context
await agent.__aenter__()
try:
    response = await agent.run("Hello")
    print(response.text)
finally:
    await agent.__aexit__(None, None, None)

Implementation Details

Entry Phase (__aenter__)

1. The method iterates over a chain of [self.client] and self.mcp_tools using itertools.chain.
2. For each element, it checks whether the object is an instance of AbstractAsyncContextManager (from contextlib).
3. If so, the object is entered into self._async_exit_stack via enter_async_context(), which calls the object's own __aenter__ and registers its __aexit__ for later cleanup.
4. Returns self to enable the as agent binding.

Exit Phase (__aexit__)

1. Calls self._async_exit_stack.aclose(), which invokes __aexit__ on every registered context manager in reverse order (LIFO).
2. Does not suppress exceptions (returns None), so any exception raised within the async with block will propagate normally after cleanup completes.

Key Dependencies

• contextlib.AsyncExitStack -- composable async context manager from the Python standard library.
• contextlib.AbstractAsyncContextManager -- abstract base class for type-checking context manager support.
• itertools.chain -- combines the client and MCP tools into a single iterable for uniform processing.
• typing_extensions.Self -- return type annotation ensuring subclasses return the correct type.

Related Pages

Implements Principle

• Principle:Microsoft_Agent_framework_Agent_Context_Management
• Heuristic:Microsoft_Agent_framework_Async_Context_Manager_Cleanup