核心 Agent 架构分析

# _agents.py:154-270
@runtime_checkable
class AgentProtocol(Protocol):
    """A protocol for an agent that can be invoked.
    
    This protocol defines the interface that all agents must implement,
    including properties for identification and methods for execution.
    
    Note:
        Protocols use structural subtyping (duck typing). Classes don't need
        to explicitly inherit from this protocol to be considered compatible.
    """
    
    id: str
    name: str | None
    description: str | None
    
    async def run(self, messages, *, thread=None, **kwargs) -> AgentResponse: ...
    
    def run_stream(self, messages, *, thread=None, **kwargs) -> AsyncIterable[AgentResponseUpdate]: ...
    
    def get_new_thread(self, **kwargs) -> AgentThread: ...

# 来自 _agents.py:167-204 的示例
class CustomAgent:
    def __init__(self):
        self.id = "custom-agent-001"
        self.name = "Custom Agent"
        self.description = "A fully custom agent implementation"

    async def run(self, messages=None, *, thread=None, **kwargs):
        return AgentResponse(messages=[], response_id="custom-response")

    def run_stream(self, messages=None, *, thread=None, **kwargs):
        async def _stream():
            yield AgentResponseUpdate()
        return _stream()

    def get_new_thread(self, **kwargs):
        return {"id": "custom-thread", "messages": []}

# 自动满足协议 - 无需继承 AgentProtocol
instance = CustomAgent()
assert isinstance(instance, AgentProtocol)  # True!

# _agents.py:275-358
class BaseAgent(SerializationMixin):
    """Base class for all Agent Framework agents.
    
    This class provides core functionality for agent implementations, including
    context providers, middleware support, and thread management.
    """
    
    def __init__(
        self,
        *,
        id: str | None = None,
        name: str | None = None,
        description: str | None = None,
        context_provider: ContextProvider | None = None,
        middleware: Sequence[Middleware] | None = None,
        additional_properties: MutableMapping[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:

# _agents.py:411-503
def as_tool(
    self,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_name: str = "task",
    arg_description: str | None = None,
    stream_callback: Callable[[AgentResponseUpdate], None]
    | Callable[[AgentResponseUpdate], Awaitable[None]]
    | None = None,
) -> FunctionTool[BaseModel, str]:

# _agents.py:384-409
def get_new_thread(self, **kwargs: Any) -> AgentThread:
    """Return a new AgentThread instance that is compatible with the agent."""
    return AgentThread(**kwargs, context_provider=self.context_provider)

async def deserialize_thread(self, serialized_thread: Any, **kwargs: Any) -> AgentThread:
    """Deserialize a thread from its serialized state."""
    thread: AgentThread = self.get_new_thread()
    await thread.update_from_thread_state(serialized_thread, **kwargs)
    return thread

async def _notify_thread_of_new_messages(
    self,
    thread: AgentThread,
    input_messages: ChatMessage | Sequence[ChatMessage],
    response_messages: ChatMessage | Sequence[ChatMessage],
    **kwargs: Any,
) -> None:
    """Notify the thread of new messages."""
    if isinstance(input_messages, ChatMessage) or len(input_messages) > 0:
        await thread.on_new_messages(input_messages)
    if isinstance(response_messages, ChatMessage) or len(response_messages) > 0:
        await thread.on_new_messages(response_messages)
    if thread.context_provider:
        await thread.context_provider.invoked(input_messages, response_messages, **kwargs)

# _agents.py:509-511
@use_agent_middleware
@use_agent_instrumentation(capture_usage=False)
class ChatAgent(BaseAgent, Generic[TOptions_co]):
    """A Chat Client Agent.
    
    This is the primary agent implementation that uses a chat client to interact
    with language models. It supports tools, context providers, middleware, and
    both streaming and non-streaming responses.
    """

# _agents.py:587-589
def __init__(
    self,
    chat_client: ChatClientProtocol[TOptions_co],
    ...
) -> None:

@runtime_checkable
class ChatClientProtocol(Protocol[TOptions_contra]):
    additional_properties: dict[str, Any]
    
    async def get_response(self, messages, *, options=None, **kwargs) -> ChatResponse: ...
    
    def get_streaming_response(self, messages, *, options=None, **kwargs) -> AsyncIterable[ChatResponseUpdate]: ...

# _agents.py:78-108
def _merge_options(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
    """Merge two options dicts, with override values taking precedence."""
    result = dict(base)
    for key, value in override.items():
        if value is None:
            continue
        if key == "tools" and result.get("tools"):
            # Combine tool lists, avoiding duplicates by name
            existing_names = {getattr(t, "name", None) for t in result["tools"]}
            unique_new = [t for t in value if getattr(t, "name", None) not in existing_names]
            result["tools"] = list(result["tools"]) + unique_new
        elif key == "logit_bias" and result.get("logit_bias"):
            # Merge logit_bias dicts
            result["logit_bias"] = {**result["logit_bias"], **value}
        elif key == "metadata" and result.get("metadata"):
            # Merge metadata dicts
            result["metadata"] = {**result["metadata"], **value}
        elif key == "instructions" and result.get("instructions"):
            # Concatenate instructions
            result["instructions"] = f"{result['instructions']}\n{value}"
        else:
            result[key] = value
    return result

# _agents.py:676-679
# We ignore the MCP Servers here and store them separately,
# we add their functions to the tools list at runtime
normalized_tools: list[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]] = (
    [] if tools_ is None else tools_ if isinstance(tools_, list) else [tools_]
)
self.mcp_tools: list[MCPTool] = [tool for tool in normalized_tools if isinstance(tool, MCPTool)]
agent_tools = [tool for tool in normalized_tools if not isinstance(tool, MCPTool)]

# Resolve final tool list (runtime provided tools + local MCP server tools)
final_tools: list[ToolProtocol | Callable[..., Any] | dict[str, Any]] = []
for tool in normalized_tools:
    if isinstance(tool, MCPTool):
        if not tool.is_connected:
            await self._async_exit_stack.enter_async_context(tool)
        final_tools.extend(tool.functions)
    else:
        final_tools.append(tool)

for mcp_server in self.mcp_tools:
    if not mcp_server.is_connected:
        await self._async_exit_stack.enter_async_context(mcp_server)
    final_tools.extend(mcp_server.functions)

# _agents.py:709-741
async def __aenter__(self) -> "Self":
    """Enter the async context manager.
    
    If any of the chat_client or local_mcp_tools are context managers,
    they will be entered into the async exit stack to ensure proper cleanup.
    """
    for context_manager in chain([self.chat_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()

async with agent:
    # MCP 连接已建立
    response = await agent.run("Hello")
# 自动清理所有连接

# _agents.py:1105-1212
def as_mcp_server(
    self,
    *,
    server_name: str = "Agent",
    version: str | None = None,
    instructions: str | None = None,
    lifespan: Callable[["Server[Any]"], AbstractAsyncContextManager[Any]] | None = None,
    **kwargs: Any,
) -> "Server[Any]":
    """Create an MCP server from an agent instance.
    
    This function automatically creates a MCP server from an agent instance...
    """

@use_agent_middleware
@use_agent_instrumentation(capture_usage=False)
class ChatAgent(BaseAgent, Generic[TOptions_co]):