diff --git a/README.md b/README.md index bc9a8b080ae..937539b95ce 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ Explore new MAF capabilities and real implementation patterns on the [official b - **Declarative Agents**: Define agents using YAML for faster setup and versioning - [Declarative agent samples](./declarative-agents/) - **Agent Skills**: Build domain-specific knowledge bases from multiple sources—files, inline code, class libraries—for agents to discover and use - - [Skills design](./docs/decisions/0037-agent-skills-design.md) + - [Skills design](./docs/decisions/0030-agent-skills-design.md) - **AF Labs**: Experimental packages for cutting-edge features including benchmarking, reinforcement learning, and research initiatives - [Labs directory](./python/packages/lab/) - **DevUI**: Interactive developer UI for agent development, testing, and debugging workflows diff --git a/docs/decisions/0026-hosted-session-identity-context.md b/docs/decisions/0026-hosted-session-identity-context.md index 9612ba3af21..e9c09b7f989 100644 --- a/docs/decisions/0026-hosted-session-identity-context.md +++ b/docs/decisions/0026-hosted-session-identity-context.md @@ -1,5 +1,5 @@ --- -status: superseded by [ADR-0030](0030-hosted-platform-context-agentserver-2.0.md) +status: superseded by [ADR-0034](0034-hosted-platform-context-agentserver-2.0.md) contact: rogerbarreto date: 2026-06-29 deciders: rogerbarreto @@ -9,7 +9,7 @@ informed: [] # Hosted session identity context for Foundry Hosting -> **Superseded by [ADR-0030](0030-hosted-platform-context-agentserver-2.0.md).** `Azure.AI.AgentServer.*` 2.0.0 (responses protocol `2.0.0`) replaced `ResponseContext.Isolation` (`UserIsolationKey` / `ChatIsolationKey`, headers `x-agent-user-isolation-key` / `x-agent-chat-isolation-key`) with `ResponseContext.PlatformContext` (`UserIdKey` / `CallId`, headers `x-agent-user-id` / `x-agent-foundry-call-id`). The chat isolation key was removed and `HostedSessionContext` is now user-only. This ADR is retained as the historical record of the original design. +> **Superseded by [ADR-0034](0034-hosted-platform-context-agentserver-2.0.md).** `Azure.AI.AgentServer.*` 2.0.0 (responses protocol `2.0.0`) replaced `ResponseContext.Isolation` (`UserIsolationKey` / `ChatIsolationKey`, headers `x-agent-user-isolation-key` / `x-agent-chat-isolation-key`) with `ResponseContext.PlatformContext` (`UserIdKey` / `CallId`, headers `x-agent-user-id` / `x-agent-foundry-call-id`). The chat isolation key was removed and `HostedSessionContext` is now user-only. This ADR is retained as the historical record of the original design. ## Context and Problem Statement diff --git a/docs/decisions/0036-structured-output.md b/docs/decisions/0029-structured-output.md similarity index 100% rename from docs/decisions/0036-structured-output.md rename to docs/decisions/0029-structured-output.md diff --git a/docs/decisions/0037-agent-skills-design.md b/docs/decisions/0030-agent-skills-design.md similarity index 100% rename from docs/decisions/0037-agent-skills-design.md rename to docs/decisions/0030-agent-skills-design.md diff --git a/docs/decisions/0038-codeact-integration.md b/docs/decisions/0031-codeact-integration.md similarity index 100% rename from docs/decisions/0038-codeact-integration.md rename to docs/decisions/0031-codeact-integration.md diff --git a/docs/decisions/0029-mcp-skill-templates-and-direct-references.md b/docs/decisions/0032-mcp-skill-templates-and-direct-references.md similarity index 100% rename from docs/decisions/0029-mcp-skill-templates-and-direct-references.md rename to docs/decisions/0032-mcp-skill-templates-and-direct-references.md diff --git a/docs/decisions/0029-python-agent-session-identity.md b/docs/decisions/0033-python-agent-session-identity.md similarity index 100% rename from docs/decisions/0029-python-agent-session-identity.md rename to docs/decisions/0033-python-agent-session-identity.md diff --git a/docs/decisions/0030-hosted-platform-context-agentserver-2.0.md b/docs/decisions/0034-hosted-platform-context-agentserver-2.0.md similarity index 100% rename from docs/decisions/0030-hosted-platform-context-agentserver-2.0.md rename to docs/decisions/0034-hosted-platform-context-agentserver-2.0.md diff --git a/docs/decisions/0031-hosted-per-user-session-storage-isolation.md b/docs/decisions/0035-hosted-per-user-session-storage-isolation.md similarity index 94% rename from docs/decisions/0031-hosted-per-user-session-storage-isolation.md rename to docs/decisions/0035-hosted-per-user-session-storage-isolation.md index beb6942c489..d18f5147dae 100644 --- a/docs/decisions/0031-hosted-per-user-session-storage-isolation.md +++ b/docs/decisions/0035-hosted-per-user-session-storage-isolation.md @@ -9,14 +9,14 @@ informed: [] # Per-agent and per-user session-storage isolation for Foundry Hosting -Builds on [ADR-0030](0030-hosted-platform-context-agentserver-2.0.md). +Builds on [ADR-0034](0034-hosted-platform-context-agentserver-2.0.md). ## Context and Problem Statement A Foundry hosted container can serve many end users (and, in .NET, many agents) over its lifetime. The `AgentSessionStore` persists each turn's `AgentSession` (which for a workflow agent carries the workflow checkpoint, and which also carries the tool-approval mapping via `ToolApprovalIdMap` in the session state -bag). [ADR-0030](0030-hosted-platform-context-agentserver-2.0.md) protected cross-user access only through +bag). [ADR-0034](0034-hosted-platform-context-agentserver-2.0.md) protected cross-user access only through the strict-resume identity check (a 403 when the persisted `HostedSessionContext.UserId` does not match the live request). The persisted artifacts themselves were keyed by `conversationId` (+ agent name), not physically partitioned per user, so a forged `conversation_id` would still resolve to another user's file @@ -34,7 +34,7 @@ hosting model. must isolate per agent as well as per user (Python hosts a single agent and needs no agent layer). - Path-traversal safety (CWE-22) for the untrusted, platform-injected user id. - Back-compat for local development (no `x-agent-user-id` header) and for direct/non-hosted store use. -- Keep the change contained and avoid the async-iterator `AsyncLocal` revert hazard from ADR-0030. +- Keep the change contained and avoid the async-iterator `AsyncLocal` revert hazard from ADR-0034. ## Considered Options @@ -45,7 +45,7 @@ hosting model. user id on the read path and yields a flat key rather than nested per-tenant directories. - An `AsyncLocal` user-context set by the handler. Rejected: the session is saved in the handler's `finally` after the streaming `yield`s, where an `AsyncLocal` set up front is reverted (the same hazard - that forced explicit call-id re-application in ADR-0030). Explicit threading is safer and clearer. + that forced explicit call-id re-application in ADR-0034). Explicit threading is safer and clearer. - A separate per-user approval store (as in Python). Rejected as unnecessary: see below. ## Decision Outcome @@ -74,7 +74,7 @@ Other elements: - Path-traversal guard: the user id is rejected (not sanitized) when it is not a single safe path segment (path separators, NUL, drive letters, rooted paths, all-dot segments). After building the path, the fully-resolved path is asserted to remain under the storage root. -- The strict-resume 403 identity check from ADR-0030 is **kept** as the second defense layer (it still +- The strict-resume 403 identity check from ADR-0034 is **kept** as the second defense layer (it still catches a session that reaches the wrong partition, e.g. via a non-partitioning custom store or in-process tampering). - **No separate approval store.** The tool-approval mapping lives in `ToolApprovalIdMap` -> @@ -94,7 +94,7 @@ Negative: - Breaking change to the experimental Foundry `AgentSessionStore` API (added `userId`). - The on-disk layout and leaf filename change (`.json` -> `c-.json`), orphaning sessions written - by the ADR-0030 release. Acceptable for an experimental package; a fresh session is created on next use. + by the ADR-0034 release. Acceptable for an experimental package; a fresh session is created on next use. ## Out of scope @@ -103,7 +103,7 @@ Negative: ## Update (2026-09-01): contract promoted to Abstractions -[ADR-0039](0039-shared-agent-session-store.md) promotes this `AgentSessionStore` contract to +[ADR-0045](0045-shared-agent-session-store.md) promotes this `AgentSessionStore` contract to `Microsoft.Agents.AI.Abstractions` and makes it the common contract for Foundry Hosting and conventional Hosting. It supersedes the required `userId` parameter with `AgentSessionStoreKey.Partitions`. Foundry Hosting adds the resolved user identity as a named partition before loading or saving a session. diff --git a/docs/decisions/0032-dotnet-hosting-protocol-helpers.md b/docs/decisions/0036-dotnet-hosting-protocol-helpers.md similarity index 99% rename from docs/decisions/0032-dotnet-hosting-protocol-helpers.md rename to docs/decisions/0036-dotnet-hosting-protocol-helpers.md index bc3d36adf38..7f728eb3f98 100644 --- a/docs/decisions/0032-dotnet-hosting-protocol-helpers.md +++ b/docs/decisions/0036-dotnet-hosting-protocol-helpers.md @@ -11,7 +11,7 @@ informed: [] Realizes the helper-first direction of [ADR-0027](0027-hosting-channels.md) for .NET. -> **Update (2026-09-01):** [ADR-0039](0039-shared-agent-session-store.md) supersedes the +> **Update (2026-09-01):** [ADR-0045](0045-shared-agent-session-store.md) supersedes the > `AgentSessionStore` portion of this decision. The protocol helper and workflow decisions remain accepted. ## Context and Problem Statement diff --git a/docs/decisions/0033-feature-usage-bitmask-user-agent.md b/docs/decisions/0037-feature-usage-bitmask-user-agent.md similarity index 100% rename from docs/decisions/0033-feature-usage-bitmask-user-agent.md rename to docs/decisions/0037-feature-usage-bitmask-user-agent.md diff --git a/docs/decisions/0034-python-session-store-serialization.md b/docs/decisions/0038-python-session-store-serialization.md similarity index 99% rename from docs/decisions/0034-python-session-store-serialization.md rename to docs/decisions/0038-python-session-store-serialization.md index 0795ac6c589..f3f67b25dbf 100644 --- a/docs/decisions/0034-python-session-store-serialization.md +++ b/docs/decisions/0038-python-session-store-serialization.md @@ -200,7 +200,7 @@ implementations. owns or re-exports `SessionStore` (this will be a breaking change in the `hosting` package). Actual `SessionStore` and `FileSessionStore` operations mark Python feature-usage index 17, -`core.session_store`, following ADR-0033's use-not-presence policy. Construction and import alone do not mark the bit. +`core.session_store`, following ADR-0037's use-not-presence policy. Construction and import alone do not mark the bit. `SessionStore` accepts opaque non-empty keys so custom backends can use their native key contracts. `FileSessionStore` accepts opaque keys up to 128 characters and encodes values that are not portable filename stems; this supports diff --git a/docs/decisions/0032-durable-azure-functions-extraction.md b/docs/decisions/0039-durable-azure-functions-extraction.md similarity index 100% rename from docs/decisions/0032-durable-azure-functions-extraction.md rename to docs/decisions/0039-durable-azure-functions-extraction.md diff --git a/docs/decisions/0035-dotnet-agent-hooks-enforcement.md b/docs/decisions/0040-dotnet-agent-hooks-enforcement.md similarity index 100% rename from docs/decisions/0035-dotnet-agent-hooks-enforcement.md rename to docs/decisions/0040-dotnet-agent-hooks-enforcement.md diff --git a/docs/decisions/0035-foundry-hosting-resilient-long-running-agents.md b/docs/decisions/0041-foundry-hosting-resilient-long-running-agents.md similarity index 100% rename from docs/decisions/0035-foundry-hosting-resilient-long-running-agents.md rename to docs/decisions/0041-foundry-hosting-resilient-long-running-agents.md diff --git a/docs/decisions/0039-python-refusal-content.md b/docs/decisions/0042-python-refusal-content.md similarity index 100% rename from docs/decisions/0039-python-refusal-content.md rename to docs/decisions/0042-python-refusal-content.md diff --git a/docs/decisions/0039-python-foundry-hosting-history-source.md b/docs/decisions/0043-python-foundry-hosting-history-source.md similarity index 100% rename from docs/decisions/0039-python-foundry-hosting-history-source.md rename to docs/decisions/0043-python-foundry-hosting-history-source.md diff --git a/docs/decisions/0040-isolate-python-lab-workspace.md b/docs/decisions/0044-isolate-python-lab-workspace.md similarity index 100% rename from docs/decisions/0040-isolate-python-lab-workspace.md rename to docs/decisions/0044-isolate-python-lab-workspace.md diff --git a/docs/decisions/0039-shared-agent-session-store.md b/docs/decisions/0045-shared-agent-session-store.md similarity index 97% rename from docs/decisions/0039-shared-agent-session-store.md rename to docs/decisions/0045-shared-agent-session-store.md index d6d33f97fca..a84432ad03f 100644 --- a/docs/decisions/0039-shared-agent-session-store.md +++ b/docs/decisions/0045-shared-agent-session-store.md @@ -92,5 +92,5 @@ Negative: ## More Information -- [ADR-0031](0031-hosted-per-user-session-storage-isolation.md) records the earlier Foundry-specific user partition. -- [ADR-0032](0032-dotnet-hosting-protocol-helpers.md) records the previous conventional Hosting contract. +- [ADR-0035](0035-hosted-per-user-session-storage-isolation.md) records the earlier Foundry-specific user partition. +- [ADR-0036](0036-dotnet-hosting-protocol-helpers.md) records the previous conventional Hosting contract. diff --git a/docs/decisions/0041-handle-provider-invalidated-responses.md b/docs/decisions/0046-handle-provider-invalidated-responses.md similarity index 100% rename from docs/decisions/0041-handle-provider-invalidated-responses.md rename to docs/decisions/0046-handle-provider-invalidated-responses.md diff --git a/docs/decisions/0043-python-mcp-runtime-context.md b/docs/decisions/0047-python-mcp-runtime-context.md similarity index 100% rename from docs/decisions/0043-python-mcp-runtime-context.md rename to docs/decisions/0047-python-mcp-runtime-context.md diff --git a/docs/decisions/0044-python-content-identity-and-protocol-metadata.md b/docs/decisions/0048-python-content-identity-and-protocol-metadata.md similarity index 98% rename from docs/decisions/0044-python-content-identity-and-protocol-metadata.md rename to docs/decisions/0048-python-content-identity-and-protocol-metadata.md index 129b50f91cb..ac0ad55a0f4 100644 --- a/docs/decisions/0044-python-content-identity-and-protocol-metadata.md +++ b/docs/decisions/0048-python-content-identity-and-protocol-metadata.md @@ -33,9 +33,9 @@ while every adapter decides placement independently. [Spec 004](../specs/004-python-function-calling-loop.md) already defines function-call identity: `function_call.Content.id` identifies one locally actionable Agent Framework occurrence, while `call_id` is the -provider's call/result correlation. [ADR 0039](0039-python-refusal-content.md) defines the flat +provider's call/result correlation. [ADR 0042](0042-python-refusal-content.md) defines the flat `model_output_kind="refusal"` marker. This decision keeps spec 004's identity semantics and the meaning of ADR -0039's marker unchanged, but moves that marker into the `core` namespace described below. It also changes which +0042's marker unchanged, but moves that marker into the `core` namespace described below. It also changes which field marks the hosted server boundary in spec 004's approval scenarios; the implementing change updates spec 004's hosted-server-boundary scenario row and its authoritative test mapping. @@ -202,7 +202,7 @@ content.additional_properties == { data under `foundry`. Each owner documents its keys in its package README and may publish a `TypedDict` for its namespace and small accessor functions. 3. **Provider-neutral data lives in `core`.** `additional_properties["core"]` holds semantics documented by - core that any package may read or write: `model_output_kind` from ADR 0039 and the caller-facing + core that any package may read or write: `model_output_kind` from ADR 0042 and the caller-facing `prompt_cache_breakpoint`. Only core adds keys to `core`. File names use the first-class `Content.name` field. 4. **Namespaces are exactly one level deep.** A namespace maps flat key names to plain JSON values (mappings, lists, strings, numbers, booleans, and `None`), never dataclass or Pydantic instances, because `Content.from_dict()` does @@ -260,7 +260,7 @@ reserved names. content that a newer provider package produced. - `model_output_kind` and `prompt_cache_breakpoint` move into `core`; their meaning is unchanged. Readers accept the flat keys indefinitely, writers write both until their next major version, and applications that set the flat keys - keep working. ADR 0039 remains the record of the refusal decision. + keep working. ADR 0042 remains the record of the refusal decision. - A package that writes namespaces requires, as its minimum core version, the core release that merges them. On an older core, streamed fragments would lose namespace fields during aggregation. - Keys used only for display, diagnostics, or a single request switch directly. @@ -286,7 +286,7 @@ nevertheless use the same namespaces; response aggregation uses the namespace-aw - Bad, because writers must replace namespaces rather than mutate them; a writer that mutates in place corrupts metadata shared with copies. The core helpers make the safe path easy but cannot enforce it. - Bad, because core gains public helper functions, and writer packages must raise their minimum core version. -- Bad, because ADR 0039's marker and the caller-facing `prompt_cache_breakpoint` move, so samples and documentation +- Bad, because ADR 0042's marker and the caller-facing `prompt_cache_breakpoint` move, so samples and documentation change, although the flat keys remain readable. - Bad, because `core` and every namespace name become reserved top-level keys. No collisions exist in this repository, but applications outside it could already use those names. @@ -373,7 +373,7 @@ forms are removed during implementation rather than documented as compatibility - [Spec 004: Python function-calling loop](../specs/004-python-function-calling-loop.md) - [Spec 005: Python content identity and protocol metadata](../specs/005-python-content-identity-and-protocol-metadata.md) -- [ADR 0039: Preserve model refusals with marked Python text content](0039-python-refusal-content.md) -- [ADR 0043: Preserve host runtime context through MCP invocation and approval](0043-python-mcp-runtime-context.md) +- [ADR 0042: Preserve model refusals with marked Python text content](0042-python-refusal-content.md) +- [ADR 0047: Preserve host runtime context through MCP invocation and approval](0047-python-mcp-runtime-context.md) - [Issue 8910: Explicit append/replace semantics for streamed content merging](https://github.com/microsoft/agent-framework/issues/8910) - [Issue 8903: Streamed code-interpreter code can be corrupted by prefix inference](https://github.com/microsoft/agent-framework/issues/8903) diff --git a/docs/decisions/0045-dotnet-unanswered-tool-approvals.md b/docs/decisions/0049-dotnet-unanswered-tool-approvals.md similarity index 100% rename from docs/decisions/0045-dotnet-unanswered-tool-approvals.md rename to docs/decisions/0049-dotnet-unanswered-tool-approvals.md diff --git a/docs/decisions/0045-python-mcp-v2-client-lifecycle.md b/docs/decisions/0050-python-mcp-v2-client-lifecycle.md similarity index 99% rename from docs/decisions/0045-python-mcp-v2-client-lifecycle.md rename to docs/decisions/0050-python-mcp-v2-client-lifecycle.md index 6e51ab6266f..279c2e40653 100644 --- a/docs/decisions/0045-python-mcp-v2-client-lifecycle.md +++ b/docs/decisions/0050-python-mcp-v2-client-lifecycle.md @@ -259,6 +259,6 @@ Committed contract tests cover: - [MCP MRTR specification](https://modelcontextprotocol.io/specification/2026-07-28/basic/patterns/mrtr) - [MCP subscriptions specification](https://modelcontextprotocol.io/specification/2026-07-28/basic/patterns/subscriptions) - [MCP caching specification](https://modelcontextprotocol.io/specification/2026-07-28/server/utilities/caching) -- [ADR 0043: MCP runtime context](0043-python-mcp-runtime-context.md) +- [ADR 0047: MCP runtime context](0047-python-mcp-runtime-context.md) This ADR was accepted through the repository ADR review process. diff --git a/docs/decisions/0048-dotnet-declarative-state-ownership.md b/docs/decisions/0051-dotnet-declarative-state-ownership.md similarity index 100% rename from docs/decisions/0048-dotnet-declarative-state-ownership.md rename to docs/decisions/0051-dotnet-declarative-state-ownership.md diff --git a/docs/decisions/0047-dotnet-workflow-owned-mcp-sessions.md b/docs/decisions/0052-dotnet-workflow-owned-mcp-sessions.md similarity index 100% rename from docs/decisions/0047-dotnet-workflow-owned-mcp-sessions.md rename to docs/decisions/0052-dotnet-workflow-owned-mcp-sessions.md diff --git a/docs/decisions/0053-python-invocation-owned-mcp-sessions.md b/docs/decisions/0053-python-invocation-owned-mcp-sessions.md new file mode 100644 index 00000000000..2c0c303f428 --- /dev/null +++ b/docs/decisions/0053-python-invocation-owned-mcp-sessions.md @@ -0,0 +1,46 @@ +--- +status: proposed +--- + +# Own declarative MCP sessions per invocation + +## Context and Problem Statement + +A shared `DefaultMCPToolHandler` caches initialized MCP clients by connection +configuration. Independently constructed workflows with identical endpoint and +headers can therefore share server-side protocol state. Workflow-owned session +identity would partition that cache, but also require Core changes for run +admission, approval routing, nested workflows, and checkpoint recovery. + +## Considered Options + +- Retain the cache and propagate workflow identity through Core and declarative requests. +- Create and close a protocol session for each invocation. +- Require every host to implement a custom handler. + +## Decision Outcome + +Choose invocation ownership, matching the lifetime principle in [.NET PR #9722](https://github.com/microsoft/agent-framework/pull/9722). +Every invocation, including `tools/list`, connects and closes its own +`MCPStreamableHTTPTool`. Cleanup completes before the call returns, including +on failure or cancellation. Cleanup failures are logged without replacing the +invocation result or error; cancellation still propagates. + +Keep the existing optional Python `client_provider` callback. Its clients are +borrowed and remain host-owned; returning `None` creates an internally owned +client. This separates HTTP connection pooling from MCP protocol-session +ownership without introducing a factory abstraction or new transport options. +Core already supports both ownership modes, so no Core changes are necessary. + +Remove the initialized-client cache, `cache_max_size`, `aclose()`, and the handler +async context manager. Hosts own admission, deadlines, cancellation, and shutdown +coordination. They must cancel and await active invocations before disposing +borrowed HTTP clients. + +This is an intentional API and behavioral break. Replace handler context +managers with ordinary construction and remove handler shutdown calls. Each call +pays connection and discovery overhead. Implicit server-session continuity no +longer survives subsequent calls, approval waits, or checkpoint resumes. +Applications needing continuity must use explicit remote state handles or an +explicitly scoped custom handler. Existing checkpoint and approval contracts +remain unchanged; no workflow identity field or checkpoint migration is added. diff --git a/docs/features/code_act/dotnet-implementation.md b/docs/features/code_act/dotnet-implementation.md index a665a8873e4..169d2eac65f 100644 --- a/docs/features/code_act/dotnet-implementation.md +++ b/docs/features/code_act/dotnet-implementation.md @@ -1,7 +1,7 @@ # CodeAct .NET implementation This document describes the .NET realization of the CodeAct design in -[`docs/decisions/0038-codeact-integration.md`](../../decisions/0038-codeact-integration.md). +[`docs/decisions/0031-codeact-integration.md`](../../decisions/0031-codeact-integration.md). This document is intentionally focused on the .NET design and public API surface. The initial public .NET type described here is `HyperlightCodeActProvider`. Future .NET backends, such as Monty, should follow the same conceptual model with their own concrete provider types rather than through a public abstract base class or a public executor parameter. @@ -23,7 +23,7 @@ Implementation-free outcome: ## What is the problem being solved? -The cross-SDK problem statement and decision rationale live in the [ADR](../../decisions/0038-codeact-integration.md). The items below narrow that statement to .NET-specific design concerns: +The cross-SDK problem statement and decision rationale live in the [ADR](../../decisions/0031-codeact-integration.md). The items below narrow that statement to .NET-specific design concerns: - Today, the easiest way to prototype CodeAct in .NET is to manually configure an `AIFunction` and wire instructions — this is fragile and requires understanding internal sandbox lifecycle details. - There is no first-class .NET design that simultaneously covers Hyperlight-backed CodeAct now, future backend-specific providers, and both tool-enabled and interpreter modes. diff --git a/docs/features/code_act/python-implementation.md b/docs/features/code_act/python-implementation.md index 6d0151c3c00..6d8a4af695b 100644 --- a/docs/features/code_act/python-implementation.md +++ b/docs/features/code_act/python-implementation.md @@ -1,7 +1,7 @@ # CodeAct Python implementation This document describes the Python realization of the CodeAct design in -[`docs/decisions/0038-codeact-integration.md`](../../decisions/0038-codeact-integration.md). +[`docs/decisions/0031-codeact-integration.md`](../../decisions/0031-codeact-integration.md). This document is intentionally focused on the Python design and public API surface. The initial public Python type described here is `HyperlightCodeActProvider`. Future Python backends, such as Monty, should follow the same conceptual model with their own concrete provider types rather than through a public abstract base class or a public executor parameter. @@ -23,7 +23,7 @@ Implementation-free outcome: ## What is the problem being solved? -The cross-SDK problem statement and decision rationale live in the [ADR](../../decisions/0038-codeact-integration.md). The items below narrow that statement to Python-specific design concerns: +The cross-SDK problem statement and decision rationale live in the [ADR](../../decisions/0031-codeact-integration.md). The items below narrow that statement to Python-specific design concerns: - Today, the easiest way to prototype CodeAct is to infer or reshape the agent's direct tool surface, which is fragile and hard to reason about. - In Python, inferring a CodeAct tool surface from generic agent tool configuration is fragile and hard to reason about. diff --git a/docs/specs/003-dotnet-hosting-protocol-helpers.md b/docs/specs/003-dotnet-hosting-protocol-helpers.md index fd98bef8be2..0e4a16cb708 100644 --- a/docs/specs/003-dotnet-hosting-protocol-helpers.md +++ b/docs/specs/003-dotnet-hosting-protocol-helpers.md @@ -9,7 +9,7 @@ informed: [] # .NET hosting: OpenAI Responses protocol helpers and optional execution state -Implements [ADR-0032](../decisions/0032-dotnet-hosting-protocol-helpers.md), which realizes the +Implements [ADR-0036](../decisions/0036-dotnet-hosting-protocol-helpers.md), which realizes the helper-first direction of [ADR-0027](../decisions/0027-hosting-channels.md) for .NET. ## What is the goal of this feature? diff --git a/docs/specs/004-feature-usage-telemetry.md b/docs/specs/004-feature-usage-telemetry.md index 3903364d3c3..96ff755083a 100644 --- a/docs/specs/004-feature-usage-telemetry.md +++ b/docs/specs/004-feature-usage-telemetry.md @@ -9,7 +9,7 @@ informed: # Feature-usage telemetry via an accumulating bitmask -> Companion design for [ADR-0033](../decisions/0033-feature-usage-bitmask-user-agent.md). +> Companion design for [ADR-0037](../decisions/0037-feature-usage-bitmask-user-agent.md). > The per-language bit tables, encoding, opt-out, and governance live in > [feature-usage-bit-registry.md](feature-usage-bit-registry.md). The registry > allocates indexes; package-local `FeatureIndex` declarations implement them. @@ -36,7 +36,7 @@ magentic / handoff), **one bit per built-in context/history provider**, selected skill source types, and separate Foundry chat/agent/memory/evals/toolbox bits (plus embedding in Python). See the -[registry](feature-usage-bit-registry.md). ADR-0033 still leaves final v1 +[registry](feature-usage-bit-registry.md). ADR-0037 still leaves final v1 granularity open. The refreshed candidate assigns 63 Python indexes and 52 .NET indexes. V1 uses 128 bits, leaving 65 Python and 76 .NET positions for additive growth. @@ -310,7 +310,7 @@ a monotonically-growing, combinatorial value must never become a metric dimension.) The version prefix leaves the door open to add it later **if** the User-Agent path cannot answer a concrete query and there is an acceptable scoped/redacted variant; v1 ships the UA path only. See -[ADR-0033 → option C](../decisions/0033-feature-usage-bitmask-user-agent.md#considered-options). +[ADR-0037 → option C](../decisions/0037-feature-usage-bitmask-user-agent.md#considered-options). ## API Changes @@ -488,8 +488,8 @@ The decision-level limitations and unresolved trade-offs — reach, per-process (not per-call) attribution, v1 granularity, fingerprinting residue, and the OTel question — are owned by the ADR (the dedicated mask-only opt-out is now decided and included). See -**[ADR-0033 → Limitations](../decisions/0033-feature-usage-bitmask-user-agent.md#limitations)** -and **[Open Questions](../decisions/0033-feature-usage-bitmask-user-agent.md#open-questions-for-decider-discussion)**. +**[ADR-0037 → Limitations](../decisions/0037-feature-usage-bitmask-user-agent.md#limitations)** +and **[Open Questions](../decisions/0037-feature-usage-bitmask-user-agent.md#open-questions-for-decider-discussion)**. This spec is the implementation reference; it does not re-litigate those choices. Implementation-only note: diff --git a/docs/specs/005-python-content-identity-and-protocol-metadata.md b/docs/specs/005-python-content-identity-and-protocol-metadata.md index 040c7528bb5..83b61c1fcf6 100644 --- a/docs/specs/005-python-content-identity-and-protocol-metadata.md +++ b/docs/specs/005-python-content-identity-and-protocol-metadata.md @@ -9,7 +9,7 @@ deciders: eavanvalkenburg ## Scope This specification defines the serialized Python contract selected by -[ADR 0044](../decisions/0044-python-content-identity-and-protocol-metadata.md). It applies to `Content`, `Message`, +[ADR 0048](../decisions/0048-python-content-identity-and-protocol-metadata.md). It applies to `Content`, `Message`, streamed response aggregation, and all Python provider and protocol adapters that emit or consume these values, including OpenAI, Foundry, Anthropic, Bedrock, Gemini, Mistral, Ollama, Claude, GitHub Copilot, Copilot Studio, Hosting Responses, Hosting A2A, DevUI, AG-UI, and A2A. diff --git a/docs/specs/feature-usage-bit-registry.md b/docs/specs/feature-usage-bit-registry.md index 65ee24303a3..211af83997a 100644 --- a/docs/specs/feature-usage-bit-registry.md +++ b/docs/specs/feature-usage-bit-registry.md @@ -1,11 +1,11 @@ # Feature-usage bit registry (per-language) -> **Status:** draft, accompanies [ADR-0033](../decisions/0033-feature-usage-bitmask-user-agent.md) +> **Status:** draft, accompanies [ADR-0037](../decisions/0037-feature-usage-bitmask-user-agent.md) > and [SPEC-004](004-feature-usage-telemetry.md). > **Version:** `1` per language · **Width:** 128-bit This document is the proposed human-readable registry for the feature-usage -mask. Until ADR-0033 is accepted and the index declarations ship, these tables +mask. Until ADR-0037 is accepted and the index declarations ship, these tables are a **candidate mapping**, not a stable wire contract. The table is the allocation authority and published decoder contract; package-local private `FeatureIndex` declarations implement the rows they own. There is no generated @@ -23,7 +23,7 @@ features were exercised** in a process — not which packages are installed. The candidate below uses package-level bits plus selected major capabilities: core agent/workflow/MCP features, stable skill source types, each orchestration pattern, each individual built-in context/history provider, and distinct Foundry -surfaces. ADR-0033 still leaves the final v1 granularity open. A feature sets its +surfaces. ADR-0037 still leaves the final v1 granularity open. A feature sets its index at first meaningful activation; the SDK shifts that index, ORs the mask, and emits the value. @@ -102,7 +102,7 @@ no \"same bit, same meaning\" rule. - **OpenTelemetry: not in v1.** Deferred primarily for privacy (a span attribute would broadcast the fingerprint into the user's general telemetry / third-party APM vendors). Left open behind the version prefix; see - [ADR-0033](../decisions/0033-feature-usage-bitmask-user-agent.md#considered-options). + [ADR-0037](../decisions/0037-feature-usage-bitmask-user-agent.md#considered-options). ## Index table — Python (`agent-framework-python`, version 1) diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/Microsoft.Agents.AI.Foundry.Hosting.csproj b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/Microsoft.Agents.AI.Foundry.Hosting.csproj index bdf92960bde..00afcb19d30 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/Microsoft.Agents.AI.Foundry.Hosting.csproj +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/Microsoft.Agents.AI.Foundry.Hosting.csproj @@ -6,7 +6,7 @@ preview Microsoft Agent Framework for Foundry Hosted Agents Provides Microsoft Agent Framework support for hosting Foundry Agents with the Azure AI Agent Service. Breaking change: this version targets the Foundry Responses container protocol v2.0 only and is not compatible with v1; use an earlier release for the v1 protocol definition. - Breaking change: this release migrates Foundry hosting to the Azure.AI.AgentServer 2.0 (Responses container protocol v2.0) surface. It is only compatible with responses protocol v2.0. Use a previous release for the v1 protocol definition. See docs/decisions/0030-hosted-platform-context-agentserver-2.0.md. + Breaking change: this release migrates Foundry hosting to the Azure.AI.AgentServer 2.0 (Responses container protocol v2.0) surface. It is only compatible with responses protocol v2.0. Use a previous release for the v1 protocol definition. See docs/decisions/0034-hosted-platform-context-agentserver-2.0.md. diff --git a/dotnet/src/Microsoft.Agents.AI.Hyperlight/README.md b/dotnet/src/Microsoft.Agents.AI.Hyperlight/README.md index f6a19d4fe51..adbb5a4de37 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hyperlight/README.md +++ b/dotnet/src/Microsoft.Agents.AI.Hyperlight/README.md @@ -1,6 +1,6 @@ # Microsoft.Agents.AI.Hyperlight -First-class [CodeAct](../../../docs/decisions/0038-codeact-integration.md) +First-class [CodeAct](../../../docs/decisions/0031-codeact-integration.md) support for the Microsoft Agent Framework, backed by the [Hyperlight](https://github.com/hyperlight-dev/hyperlight) VM-isolated sandbox. diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index b66484a40ec..245f2051cb4 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -234,7 +234,7 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID ### Model Context Protocol (`_mcp.py`) -- **Client lifecycle design** - [ADR 0045](../../../docs/decisions/0045-python-mcp-v2-client-lifecycle.md) +- **Client lifecycle design** - [ADR 0050](../../../docs/decisions/0050-python-mcp-v2-client-lifecycle.md) defines the ownership boundary. Both framework-created and caller-supplied `_ClientMCPConnection` instances expose and route operations through a high-level `mcp.Client`; their `session` remains `None`, and `is_framework_owned` determines whether Agent Framework may manage the Client lifecycle. Caller-supplied diff --git a/python/packages/declarative/AGENTS.md b/python/packages/declarative/AGENTS.md index 1c725aff316..a52064c2ea8 100644 --- a/python/packages/declarative/AGENTS.md +++ b/python/packages/declarative/AGENTS.md @@ -23,20 +23,25 @@ series because response caching and default-header integration use its internals ## MCP Handler Lifetimes -`DefaultMCPToolHandler` caches/coalesces sessions only without a `client_provider`. -With a provider, every invocation (including `tools/list`) gets a fresh tool/session, -even if the provider returns `None` or a shared HTTP client. Invocation cleanup closes -the session and any internally owned fallback client, never caller-owned HTTP clients. +`DefaultMCPToolHandler` creates a fresh tool/session for every invocation, including +`tools/list`, with or without a `client_provider`. A provider may return `None` or a +shared HTTP client without sharing protocol sessions. Invocation cleanup closes +the session and any internally owned HTTP client, never caller-owned HTTP clients. The handler forwards its configurable `max_discovery_pages` and `max_discovery_items` limits to every internally created `MCPStreamableHTTPTool`. -Shutdown waits for active provider-backed invocations to clean up. Calling `aclose()` -from an active invocation's context (including inherited child tasks and cleanup) -raises `RuntimeError` before changing handler state; shutdown must run outside that -context or after invocation completion. - -Per-invocation sessions intentionally add connection overhead and lose server -session continuity between invocations; -shared session ownership requires an explicitly scoped custom `MCPToolHandler`. +The handler has no cache, `cache_max_size`, `aclose()`, or async context manager. +Hosts own admission, deadlines, cancellation, and shutdown of active calls; cancel +and await them before closing any borrowed HTTP client. + +Provider, connection, discovery, and transport failures propagate and fail the +action, including after approval. Remote tool error responses remain +`MCPToolResult(is_error=True)`; executors do not convert raised exceptions into results. + +Per-invocation sessions add connection overhead and intentionally lose implicit +server session continuity across calls, approval waits, and checkpoint resumes. +Applications requiring continuity must use explicit remote state handles or an +explicitly scoped custom `MCPToolHandler`. No Core workflow scope or checkpoint +migration is involved. ## MCP Approval Context diff --git a/python/packages/declarative/README.md b/python/packages/declarative/README.md index 0a4b28549d1..652d03b30ff 100644 --- a/python/packages/declarative/README.md +++ b/python/packages/declarative/README.md @@ -72,22 +72,36 @@ on HTTPX internals. Compatibility tests cover HTTPX 0.27.0 and 0.28.1. ## MCP handler session lifetime -`DefaultMCPToolHandler()` reuses MCP sessions through a bounded LRU cache. -When configured with `client_provider`, it instead invokes the provider and -creates a fresh MCP tool/session for every invocation, including `tools/list`. -Sessions are never reused in this mode, even if the provider returns `None` -or the same `httpx.AsyncClient` instance. +**Breaking change:** `DefaultMCPToolHandler` creates and closes a fresh MCP +tool/session for every invocation, including `tools/list`. Independently +constructed workflows may safely share a handler without sharing MCP protocol +state, even with identical endpoint and header configuration. There is no +initialized-client or discovery cache. Each invocation closes its tool/session on success, failure, or cancellation. -Internally created fallback HTTP clients are also closed; caller-supplied HTTP -clients remain caller-owned and are never closed by the handler. Calling -`aclose()` rejects new invocations and waits for active provider-backed -invocations to finish cleaning up. Calling it from an active invocation's -context (including provider callbacks, inherited child tasks, and cleanup) -raises `RuntimeError` before changing handler state, rather than waiting on -itself. Close the handler outside that context or after the invocation completes. - -This intentionally incurs connection and initialization overhead and does not -retain server session state between provider-backed invocations. Applications -that require shared session ownership must implement an explicitly scoped -custom `MCPToolHandler`; there is no provider-backed session-cache opt-in. +Internally created HTTP clients are also closed. The optional `client_provider` +is called once per invocation and may return a reusable `httpx.AsyncClient` +to share an HTTP connection pool, not an MCP session. Returned clients are +borrowed and never closed by the handler. Returning `None` selects an internally +owned client. + +Provider, connection, discovery, and transport failures raise exceptions and fail +the workflow action, including after approval. Remote tool error responses still +produce `MCPToolResult(is_error=True)` and populate `output.result` with an error +string. Custom handlers should return error results only for tool-level errors; +raised exceptions are not converted into workflow results. + +The handler no longer accepts `cache_max_size` or implements `aclose()` or an +async context manager. Replace `async with DefaultMCPToolHandler() as handler:` +with `handler = DefaultMCPToolHandler()` and remove handler shutdown calls. +Continue to manage caller-owned HTTP clients with their own context manager. +Hosts must cancel and await active invocations before closing borrowed clients; +admission limits and deadlines also belong to the host. + +This intentionally adds connection and initialization overhead. Implicit +server-side state does not persist across tool calls, approval waits, or +checkpoint resumes. Workflow checkpoints and approval verification are +unchanged; no request-scope migration flag is required. Applications that need +MCP session continuity must provide an explicitly scoped custom `MCPToolHandler` +or pass explicit remote state handles as tool arguments. See the +[decision record](../../../docs/decisions/0053-python-invocation-owned-mcp-sessions.md). diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_mcp.py b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_mcp.py index 28ff521b0c4..3d4e059a896 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_mcp.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_mcp.py @@ -23,6 +23,8 @@ authors must trust the MCP server they invoke. """ +from __future__ import annotations + import hmac import json import logging @@ -32,7 +34,6 @@ from dataclasses import dataclass, field from typing import Any -import httpx2 as httpx from agent_framework import ( Content, Message, @@ -40,7 +41,6 @@ handler, response_handler, ) -from agent_framework.exceptions import ToolExecutionException from ._declarative_base import ( ActionComplete, @@ -177,6 +177,8 @@ class InvokeMcpToolActionExecutor(DeclarativeActionExecutor): - On error returned by the handler (``is_error=True``): assigns ``"Error: "`` to ``output.result`` and completes normally (parity with .NET ``AssignErrorAsync``). + - Setup, transport, and protocol exceptions propagate and fail the action, + including when resuming an approved invocation. .. note:: @@ -243,7 +245,7 @@ async def handle_action( ) return - result = await self._invoke_with_narrow_catch(invocation) + result = await self._mcp_tool_handler.invoke_tool(invocation) await self._process_result( ctx=ctx, state=state, @@ -357,7 +359,7 @@ async def handle_approval_response( ) await self._request_approval(invocation, dict(metadata), ctx) return - result = await self._invoke_with_narrow_catch(invocation) + result = await self._mcp_tool_handler.invoke_tool(invocation) await self._process_result( ctx=ctx, state=state, @@ -472,49 +474,6 @@ def _get_auto_send(self, state: DeclarativeWorkflowState) -> bool: return evaluated.strip().lower() in {"true", "1", "yes"} return bool(evaluated) - # ----- Invocation + error handling ---------------------------------------- - - async def _invoke_with_narrow_catch(self, invocation: MCPToolInvocation) -> MCPToolResult: - """Invoke the handler with a narrow exception catch. - - Only known transport / tool exceptions are normalised to an error - result. Programmer bugs (TypeError, ValueError from misuse, etc.) - propagate so they fail loudly. - - ``asyncio.CancelledError`` is a ``BaseException``, not ``Exception``, - so it is not caught here and propagates unchanged for workflow - cancellation. - """ - try: - return await self._mcp_tool_handler.invoke_tool(invocation) - except ToolExecutionException as exc: - message = str(exc) or type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - except httpx.HTTPError as exc: - message = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - except Exception as exc: - try: - from mcp.shared.exceptions import MCPError - except ImportError: # pragma: no cover - mcp is a hard dep - raise - if isinstance(exc, MCPError): - message = str(exc) or type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - raise - # ----- Result handling ----------------------------------------------------- async def _process_result( diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_mcp_handler.py b/python/packages/declarative/agent_framework_declarative/_workflows/_mcp_handler.py index f7ec768a6a9..b4a8c18b9d9 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_mcp_handler.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_mcp_handler.py @@ -1,46 +1,27 @@ # Copyright (c) Microsoft. All rights reserved. -"""MCP tool handler abstraction for declarative workflows. - -Mirrors the .NET ``IMcpToolHandler`` / ``DefaultMcpToolHandler`` pair from -``Microsoft.Agents.AI.Workflows.Declarative.Mcp``. Provides: - -- :class:`MCPToolInvocation` — request input data passed from the executor. -- :class:`MCPToolResult` — response data returned to the executor. -- :class:`MCPToolHandler` — :class:`typing.Protocol` callers implement to plug - in custom transports (e.g. with allowlisting, Foundry connection resolution, - per-server auth, etc.). -- :class:`DefaultMCPToolHandler` — production-grade default backed by - :class:`agent_framework.MCPStreamableHTTPTool`. - -Security note: :class:`DefaultMCPToolHandler` performs **no** URL filtering or -SSRF protection. Production deployments should supply a custom handler that -enforces an allowlist or DNS-rebinding-resistant policy. This split mirrors the -.NET design. - -Prompt-injection note: MCP tool outputs flow back into agent conversations -(via ``conversationId`` and Tool-role messages emitted by the executor) so -they share the same risk surface as ``HttpRequestAction``. Workflow authors -must trust the MCP server they invoke. +"""MCP tool handlers for declarative workflows. + +The default handler owns a separate MCP protocol session for each invocation. +Custom handlers may implement explicitly scoped session continuity or additional +transport policy. Neither the default handler nor this protocol filters URLs: +hosts must enforce their own destination allowlists and trust the MCP servers +whose outputs enter workflow conversations. """ from __future__ import annotations import asyncio -import contextlib -import hashlib import json import logging -from collections import OrderedDict from collections.abc import Awaitable, Callable -from contextvars import ContextVar, Token from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, ClassVar, Protocol, cast, runtime_checkable +from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable import httpx2 as httpx if TYPE_CHECKING: - from agent_framework import Content + from agent_framework import Content, MCPStreamableHTTPTool __all__ = [ "ClientProvider", @@ -52,28 +33,16 @@ logger = logging.getLogger(__name__) -_DEFAULT_CACHE_MAX_SIZE = 32 - @dataclass class MCPToolInvocation: - """Description of an MCP tool call to be dispatched by a :class:`MCPToolHandler`. - - Mirrors the input parameters of the .NET ``IMcpToolHandler.InvokeToolAsync`` - method. Field semantics: - - - ``server_url``: Absolute URL of the MCP server. Already evaluated from - the YAML expression. - - ``server_label``: Optional human-readable label used for diagnostics - and as the underlying ``MCPStreamableHTTPTool`` name. - - ``tool_name``: Name of the tool to invoke on the MCP server. - - ``arguments``: Tool arguments. Already evaluated; values may be any - JSON-serialisable Python object (str, int, bool, dict, list, None). - - ``headers``: Outbound HTTP headers (e.g. authentication). Empty values - are skipped by the executor before construction. - - ``connection_name``: Optional Foundry connection name forwarded for - handlers that resolve auth/credentials by connection. The default - handler does not consume this field. + """Evaluated input for an MCP tool call. + + ``server_url`` identifies the server and ``server_label`` names its client. + ``tool_name`` is either a remote tool or the reserved ``tools/list`` operation. + ``arguments`` contains JSON-serialisable tool arguments; ``headers`` contains + evaluated outbound headers. ``connection_name`` is available to custom + handlers and client providers that resolve host-owned connections. """ server_url: str @@ -84,337 +53,150 @@ class MCPToolInvocation: connection_name: str | None = None -def _empty_outputs() -> list[Any]: - """Default factory for ``MCPToolResult.outputs``. - - Typed as ``list[Any]`` here to keep the dataclass field's runtime - factory simple; the public type on :class:`MCPToolResult` is - ``list[Content]``. - """ - return [] - - @dataclass class MCPToolResult: - """Response returned by an :class:`MCPToolHandler`. + """Parsed MCP output, or a remote tool error the workflow can store in its result.""" - Mirrors the .NET ``McpServerToolResultContent`` shape. ``outputs`` is a - list of :class:`agent_framework.Content` items as parsed by the MCP - transport (TextContent / DataContent / UriContent / etc.). - - On error, ``is_error`` is ``True``, ``error_message`` carries a human - readable description, and ``outputs`` typically contains a single - ``Content.from_text("Error: ...")`` entry for downstream display. - """ - - outputs: list[Content] = field(default_factory=_empty_outputs) + outputs: list[Content] = field(default_factory=list) # pyright: ignore[reportUnknownVariableType] is_error: bool = False error_message: str | None = None @runtime_checkable class MCPToolHandler(Protocol): - """Protocol for MCP tool handlers used by ``InvokeMcpTool``. + """Transport contract for declarative ``InvokeMcpTool`` actions. - Mirrors :class:`HttpRequestHandler` — declares ONLY the invocation method. - Lifecycle methods (``aclose`` / ``__aenter__`` / ``__aexit__``) are NOT - part of the Protocol; concrete implementations may add them as - appropriate. - - Implementations must be safe to call concurrently from multiple workflow - runs. Implementations are responsible for any URL allowlisting, SSRF - guards, retry policies, auth resolution, and other policies the workflow - author wants applied. + Implementations must support concurrent calls from independent workflows. + They own destination validation, authentication, retries, and any explicitly + scoped protocol-session continuity. Lifecycle methods are not part of this + protocol. """ async def invoke_tool(self, invocation: MCPToolInvocation) -> MCPToolResult: - """Dispatch ``invocation`` and return the result. - - Args: - invocation: Description of the MCP tool call to perform. - - Returns: - The :class:`MCPToolResult` carrying the parsed outputs (or an - error flag if the tool raised). Implementations SHOULD return a - result with ``is_error=True`` rather than raising for transport - or tool-level failures, so the workflow can store the message in - ``output.result`` (matching .NET ``AssignErrorAsync`` behaviour). - They MAY raise on unexpected programming errors — these will be - propagated unchanged by the executor so they fail loudly. - """ + """Return tool output or a remote tool error; propagate setup and transport failures.""" ... ClientProvider = Callable[[MCPToolInvocation], Awaitable["httpx.AsyncClient | None"]] -@dataclass -class _CacheEntry: - """Internal record stored in the LRU cache.""" +class DefaultMCPToolHandler: + """Invoke MCP tools with a separate protocol client for every call. - tool: Any # MCPStreamableHTTPTool — typed Any to avoid import at module load - owned_httpx_client: httpx.AsyncClient | None + Every invocation, including ``tools/list``, connects and closes its own + :class:`agent_framework.MCPStreamableHTTPTool`. No initialized clients, + session state, credentials, or discovery inventories are cached here. + The optional ``client_provider`` is called for every invocation. Its HTTP + client is borrowed and never closed by this handler, so the host may reuse + HTTP connection pools without sharing MCP sessions. Returning ``None`` uses + an internally owned HTTP client that is closed with the invocation. -class DefaultMCPToolHandler: - """Default :class:`MCPToolHandler` backed by :class:`agent_framework.MCPStreamableHTTPTool`. - - Without a ``client_provider``, caches one - :class:`agent_framework.MCPStreamableHTTPTool` instance per - ``(server_url, server_label, connection_name, headers_hash)`` in a bounded - LRU. The cache prevents re-establishing an MCP session for every - invocation while ensuring different header sets (auth tokens) cannot - share a session — matches the .NET design intent while bounding - cardinality. ``server_label`` and ``connection_name`` also participate - in the key to distinguish logical connections. - Header *names* are lower-cased inside the hash payload only — the - headers passed on the wire keep the caller's original casing — so two - YAML actions that spell ``Authorization`` differently still share a - cache entry. - - Construction modes: - - 1. ``DefaultMCPToolHandler()`` — owns its own ``httpx.AsyncClient`` - instances created lazily per cache entry. Closed by :meth:`aclose`. - 2. ``DefaultMCPToolHandler(client_provider=cb)`` — per-invocation client - lookup (parity with .NET ``httpClientProvider`` callback). The - callback receives the full :class:`MCPToolInvocation` so it can - dispatch on ``server_url`` / ``connection_name`` / ``server_label``. - The callback is invoked for every call, including ``tools/list``. - Each call creates and closes its own MCP tool/session, even when the - callback returns ``None`` or the same client object. Returning ``None`` - falls back to an internally-created client, closed with the invocation. - Caller-supplied clients are never closed by this handler. - - This intentionally trades connection/session reuse for invocation - isolation. Server session state is not retained across invocations; - callers needing shared session ownership must implement an explicitly - scoped custom :class:`MCPToolHandler`. - - .. warning:: - - This handler performs **no** URL filtering or SSRF protection. Wrap - or replace it with a custom handler in production deployments. - - Args: - client_provider: Optional per-invocation ``httpx.AsyncClient`` provider. - cache_max_size: Maximum number of cached MCP clients in no-provider mode. - When exceeded, the least-recently-used entry is evicted and its - owned client closed. Defaults to ``32``. Does not enable session - caching when a provider is configured. - max_discovery_pages: Maximum pages accepted independently for each tool catalog fetch. + The handler owns no shared disposable resources and has no shutdown method. + Hosts must cancel and await active invocations before closing borrowed HTTP + clients. Admission limits and invocation deadlines are also host-owned. + Implicit server-side state does not survive later calls, approval waits, or + checkpoint resumes. Use a custom scoped handler or explicit remote state + handles when continuity is required. + + This handler performs no URL filtering or SSRF protection. + + Keyword Args: + client_provider: Optional per-invocation provider of a borrowed HTTP client. + max_discovery_pages: Maximum pages accepted per tool catalog fetch. Defaults to 100; ``None`` disables the limit. - max_discovery_items: Maximum tools accepted across each tool catalog fetch. + max_discovery_items: Maximum tools accepted per tool catalog fetch. Defaults to 1,000; ``None`` disables the limit. """ LIST_TOOLS_TOOL_NAME: ClassVar[str] = "tools/list" - """Reserved ``tool_name`` that maps an :class:`MCPToolHandler` invocation - to the MCP protocol ``tools/list`` discovery operation. - - The constant matches the underlying MCP method name so a single - string travels unchanged through host code, YAML, and the protocol - wire. When this handler receives an invocation with this name it - pages through ``session.list_tools()`` and returns the catalog as a - single ``TextContent`` containing JSON of shape - ``{"tools": [{name, description, inputSchema, outputSchema}, ...]}``. - Workflows can reference this name from an ``InvokeMcpTool`` declarative - action to introspect a server's tool surface without an extra round-trip - from host code. + """Reserved operation returning a JSON catalog with a ``tools`` array. + + Each item contains ``name``, ``description``, ``inputSchema``, and + ``outputSchema`` in that order. The catalog is paginated within one + invocation, not retained for subsequent calls. """ def __init__( self, *, client_provider: ClientProvider | None = None, - cache_max_size: int = _DEFAULT_CACHE_MAX_SIZE, max_discovery_pages: int | None = 100, max_discovery_items: int | None = 1_000, ) -> None: - if cache_max_size <= 0: - raise ValueError(f"cache_max_size must be positive, got {cache_max_size}") if max_discovery_pages is not None and max_discovery_pages <= 0: raise ValueError("max_discovery_pages must be positive or None.") if max_discovery_items is not None and max_discovery_items <= 0: raise ValueError("max_discovery_items must be positive or None.") self._client_provider = client_provider - self._cache_max_size = cache_max_size self._max_discovery_pages = max_discovery_pages self._max_discovery_items = max_discovery_items - self._cache: OrderedDict[tuple[str, str, str, str], _CacheEntry] = OrderedDict() - # Outer lock guards the cache + in-flight-future map only — never - # held across network I/O. - self._cache_lock = asyncio.Lock() - # Per-key in-flight futures: while one task is connecting, other - # tasks awaiting the same key will await the same future and share - # the resulting cache entry. - self._inflight: dict[tuple[str, str, str, str], asyncio.Future[_CacheEntry]] = {} - # Completion signals only: provider-backed calls never share entries. - self._active_invocations: set[asyncio.Future[None]] = set() - # Keep ancestry so a completed nested call cannot hide an active parent - # in the context inherited by child tasks, including cleanup tasks. - self._invocation_context: ContextVar[tuple[asyncio.Future[None], ...]] = ContextVar( - f"default_mcp_tool_handler_invocations_{id(self)}", default=() - ) - # Set by ``aclose`` to prevent post-close cache insertions and to - # reject new ``invoke_tool`` calls. Once set, never cleared. - self._closed = False async def invoke_tool(self, invocation: MCPToolInvocation) -> MCPToolResult: - """Invoke ``invocation.tool_name`` on an MCP client for the server. - - The reserved name :attr:`LIST_TOOLS_TOOL_NAME` (``"tools/list"``) is - intercepted client-side: instead of being forwarded as a tool call, - it is translated to an MCP ``session.list_tools()`` discovery - operation (paginated automatically) and returned as a single - ``TextContent`` containing a JSON tool catalog. - """ - from agent_framework import Content + """Connect, dispatch, and finish protocol cleanup before returning.""" + from agent_framework import MCPStreamableHTTPTool - # Reserved-name args validation runs before connect: rejecting bad - # input shouldn't require establishing an MCP session. if invocation.tool_name == self.LIST_TOOLS_TOOL_NAME and invocation.arguments: message = f"The reserved MCP '{self.LIST_TOOLS_TOOL_NAME}' operation does not accept tool arguments." - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) + raise ValueError(message) - entry: _CacheEntry | None = None - completion: asyncio.Future[None] | None = None - context_token: Token[tuple[asyncio.Future[None], ...]] | None = None + tool: MCPStreamableHTTPTool | None = None try: - try: - if self._client_provider is None: - entry = await self._get_or_create_entry(invocation) - else: - async with self._cache_lock: - if self._closed: - raise RuntimeError("DefaultMCPToolHandler is closed") - completion = asyncio.get_running_loop().create_future() - self._active_invocations.add(completion) - context_token = self._invocation_context.set((*self._invocation_context.get(), completion)) - entry = await self._create_entry(invocation) - if self._closed: - raise RuntimeError("DefaultMCPToolHandler is closed") - except Exception as exc: - # Connect / cache lookup failures surface as tool errors so the - # workflow can store them at output.result without crashing. - logger.warning( - "DefaultMCPToolHandler: failed to obtain MCP client for url=%s tool=%s: %s", - invocation.server_url, - invocation.tool_name, - exc, - ) - message = f"Failed to connect to MCP server: {type(exc).__name__}: {exc}".rstrip(": ") - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - - return await self._invoke_entry(entry, invocation) + client = await self._client_provider(invocation) if self._client_provider is not None else None + tool = MCPStreamableHTTPTool( + name=invocation.server_label or "McpClient", + url=invocation.server_url, + load_prompts=False, + http_client=client, + static_headers=dict(invocation.headers), + max_discovery_pages=self._max_discovery_pages, + max_discovery_items=self._max_discovery_items, + ) + await tool.connect() + return await self._invoke_connected_tool(tool, invocation) finally: - if completion is not None: - try: - if entry is not None: - await self._close_invocation_entry(entry) - finally: - if context_token is not None: - self._invocation_context.reset(context_token) - self._active_invocations.discard(completion) - completion.set_result(None) - - async def _invoke_entry(self, entry: _CacheEntry, invocation: MCPToolInvocation) -> MCPToolResult: - """Dispatch a connected invocation and preserve tool error mapping.""" + if tool is not None: + await self._close_tool(tool) + + async def _invoke_connected_tool(self, tool: MCPStreamableHTTPTool, invocation: MCPToolInvocation) -> MCPToolResult: from agent_framework import Content from agent_framework.exceptions import ToolExecutionException + if invocation.tool_name == self.LIST_TOOLS_TOOL_NAME: + return await self._invoke_list_tools(tool) try: - if invocation.tool_name == self.LIST_TOOLS_TOOL_NAME: - return await self._invoke_list_tools(entry) - raw = await entry.tool.call_tool(invocation.tool_name, **invocation.arguments) + raw = await tool.call_tool(invocation.tool_name, **invocation.arguments) except ToolExecutionException as exc: + # Core chains protocol/transport failures, unlike remote is_error responses. + if exc.__cause__ is not None: + raise logger.info( "DefaultMCPToolHandler: tool '%s' on '%s' raised ToolExecutionException", invocation.tool_name, invocation.server_url, ) message = str(exc) or type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - except httpx.HTTPError as exc: - message = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - except Exception as exc: - # Be defensive about MCP errors that may bubble up without being - # wrapped in ToolExecutionException by custom parsers. - try: - from mcp.shared.exceptions import MCPError - except ImportError: # pragma: no cover - mcp is a hard dep but stay defensive - raise - if isinstance(exc, MCPError): - message = str(exc) or type(exc).__name__ - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - raise - - # Defensive normalisation: call_tool is typed ``str | list[Content]``. - # Default parser returns list, but custom parse_tool_results may return str. - if isinstance(raw, str): - outputs: list[Content] = [Content.from_text(raw)] - else: - outputs = list(raw) + return MCPToolResult(outputs=[Content.from_text(f"Error: {message}")], is_error=True, error_message=message) + + outputs = [Content.from_text(raw)] if isinstance(raw, str) else list(raw) return MCPToolResult(outputs=outputs) @staticmethod - async def _invoke_list_tools(entry: _CacheEntry) -> MCPToolResult: - """Handle the reserved :attr:`LIST_TOOLS_TOOL_NAME` invocation. - - Pages through the normalized MCP connection (mirroring the pagination loop - in :meth:`agent_framework.MCPTool.load_tools`) and serialises the - full catalog as a single ``TextContent`` containing JSON of shape - ``{"tools": [{name, description, inputSchema, outputSchema}, ...]}``. - - The output shape, property names, and property order are stable so - downstream PowerFx expressions can rely on the schema. ``indent=2`` - produces human-readable JSON for the conversation log; - ``allow_nan=False`` guards against producing non-conformant JSON - ``NaN``/``Infinity`` tokens if a misbehaving server returns such - values in a schema. - """ + async def _invoke_list_tools(tool: MCPStreamableHTTPTool) -> MCPToolResult: + """Preserve the catalog shape and pagination used by workflow expressions.""" from agent_framework import Content + from mcp import types as mcp_types - require_connection: Any = getattr(entry.tool, "_require_connection", None) + require_connection: Any = getattr(tool, "_require_connection", None) connection: Any | None = None session: Any | None = None if callable(require_connection): - with contextlib.suppress(RuntimeError): - connection = require_connection() + connection = require_connection() else: - session = getattr(entry.tool, "session", None) + session = getattr(tool, "session", None) if connection is None and session is None: - message = "MCP session is not connected; cannot list tools." - return MCPToolResult( - outputs=[Content.from_text(f"Error: {message}")], - is_error=True, - error_message=message, - ) - - # Lazy import keeps ``mcp`` types out of module import time. - from mcp import types as mcp_types + raise RuntimeError("MCP session is not connected; cannot list tools.") collected: list[Any] = [] params: mcp_types.PaginatedRequestParams | None = None @@ -434,267 +216,29 @@ async def _invoke_list_tools(entry: _CacheEntry) -> MCPToolResult: payload = { "tools": [ { - "name": tool.name, - "description": tool.description, - "inputSchema": getattr(tool, "input_schema", getattr(tool, "inputSchema", None)), - "outputSchema": getattr(tool, "output_schema", getattr(tool, "outputSchema", None)), + "name": item.name, + "description": item.description, + "inputSchema": getattr(item, "input_schema", getattr(item, "inputSchema", None)), + "outputSchema": getattr(item, "output_schema", getattr(item, "outputSchema", None)), } - for tool in collected - ], + for item in collected + ] } return MCPToolResult(outputs=[Content.from_text(json.dumps(payload, indent=2, allow_nan=False))]) - async def aclose(self) -> None: - """Close cached clients and wait for provider-backed invocations to clean up. - - Caller-supplied :class:`httpx.AsyncClient` instances (returned by the - ``client_provider`` callback) are NOT closed. - - Provider-backed calls already executing may finish; pending connections - are rejected after connecting and cleaned up by their invocation. - Concurrent shutdown calls wait for those invocations as well. - - Idempotent. In no-provider mode, a second call returns immediately. - Drains any in-flight ``_create_entry`` tasks before returning so their resources are - cleaned up; the in-flight tasks see ``self._closed`` in phase 3 of - :meth:`_get_or_create_entry`, close their own entry, and resolve - their future with ``RuntimeError("DefaultMCPToolHandler is closed")``. - - Raises: - RuntimeError: If called from an active provider-backed invocation's - context, including inherited child tasks and cleanup. Rejected - before changing handler state to avoid waiting on itself. - """ - if any(not completion.done() for completion in self._invocation_context.get()): - raise RuntimeError( - "DefaultMCPToolHandler.aclose() cannot be called from an active provider-backed invocation" - ) - async with self._cache_lock: - if self._closed and self._client_provider is None: - return - self._closed = True - entries = list(self._cache.values()) - self._cache.clear() - inflight_futures = list(self._inflight.values()) - active_invocations = list(self._active_invocations) - - for completion in active_invocations: - # Cancelling shutdown must not cancel an invocation's completion signal. - await asyncio.shield(completion) - - # Wait for in-flight creations to finish their self-cleanup. Each - # in-flight task self-closes its entry under the closed-flag branch - # in phase 3 and resolves its future with ``RuntimeError``; we - # swallow it here because the failure is expected at shutdown. - for fut in inflight_futures: - try: - await fut - except BaseException: - logger.debug("DefaultMCPToolHandler: in-flight future raised during aclose", exc_info=True) - continue - - for entry in entries: - await self._close_entry(entry) - - async def __aenter__(self) -> DefaultMCPToolHandler: - return self - - async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: - await self.aclose() - - # ------------------------------------------------------------------ - # Internal helpers - # ------------------------------------------------------------------ - - async def _get_or_create_entry(self, invocation: MCPToolInvocation) -> _CacheEntry: - """Look up (or create) the cached MCP client for this invocation.""" - key = self._cache_key( - invocation.server_url, - invocation.server_label, - invocation.connection_name, - invocation.headers, - ) - - # Phase 1: check the cache and either claim creation or wait for an - # already in-flight creation. - creating = False - async with self._cache_lock: - if self._closed: - raise RuntimeError("DefaultMCPToolHandler is closed") - existing = self._cache.get(key) - if existing is not None: - self._cache.move_to_end(key) - return existing - inflight = self._inflight.get(key) - if inflight is None: - inflight = asyncio.get_running_loop().create_future() - self._inflight[key] = inflight - creating = True - - if not creating: - return await inflight - - # Phase 2: we own creation. Build the entry outside the lock. - try: - entry = await self._create_entry(invocation) - except BaseException as exc: - async with self._cache_lock: - self._inflight.pop(key, None) - if not inflight.done(): - inflight.set_exception(exc if isinstance(exc, BaseException) else RuntimeError(str(exc))) - # Mark the exception retrieved to suppress noisy "Future exception - # was never retrieved" warnings when there are no other awaiters - # (other awaiters still see the exception through their ``await``). - inflight.exception() - raise - - # Phase 3: insert with LRU eviction; resolve the in-flight future. - # If ``aclose`` ran while we were connecting, ``_closed`` is now - # True; don't insert into the cache (it has been drained), close - # the just-built entry, and surface the closed-handler error to - # all awaiters of the future. - evicted: _CacheEntry | None = None - duplicate: _CacheEntry | None = None - handler_closed = False - async with self._cache_lock: - self._inflight.pop(key, None) - if self._closed: - handler_closed = True - else: - existing = self._cache.get(key) - if existing is not None: - # Another writer beat us; prefer the existing entry and - # discard ours after the lock is released. - self._cache.move_to_end(key) - duplicate = entry - entry = existing - else: - self._cache[key] = entry - self._cache.move_to_end(key) - if len(self._cache) > self._cache_max_size: - _evicted_key, evicted = self._cache.popitem(last=False) - if not inflight.done(): - inflight.set_result(entry) - - if handler_closed: - # Close our orphaned entry; resolve the future with a clear - # error so the caller (and any other awaiters) surface a - # consistent "handler is closed" failure rather than receiving - # an entry we are about to close behind their back. - await self._close_entry(entry) - err = RuntimeError("DefaultMCPToolHandler is closed") - if not inflight.done(): - inflight.set_exception(err) - inflight.exception() - raise err - if duplicate is not None: - await self._close_entry(duplicate) - if evicted is not None: - await self._close_entry(evicted) - return entry - - async def _create_entry(self, invocation: MCPToolInvocation) -> _CacheEntry: - """Construct (and connect) a fresh MCP client for ``invocation``.""" - from agent_framework import MCPStreamableHTTPTool - - provided_client: httpx.AsyncClient | None = None - if self._client_provider is not None: - provided_client = await self._client_provider(invocation) - if self._closed: - raise RuntimeError("DefaultMCPToolHandler is closed") - # Capture headers for this entry so the header_provider closure - # always returns the same set, regardless of the runtime kwargs. - captured_headers = dict(invocation.headers) - - def _header_provider(_kwargs: dict[str, Any]) -> dict[str, str]: - return captured_headers - - tool: Any = MCPStreamableHTTPTool( - name=invocation.server_label or "McpClient", - url=invocation.server_url, - load_prompts=False, - http_client=provided_client, - header_provider=_header_provider if captured_headers else None, - max_discovery_pages=self._max_discovery_pages, - max_discovery_items=self._max_discovery_items, - ) - try: - await tool.connect() - except BaseException: - failed_entry = _CacheEntry( - tool=tool, - owned_httpx_client=( - cast("httpx.AsyncClient | None", getattr(tool, "_httpx_client", None)) - if provided_client is None - else None - ), - ) - if self._client_provider is not None: - await self._close_invocation_entry(failed_entry) - else: - await self._close_entry(failed_entry) - raise - - # ``MCPStreamableHTTPTool.get_mcp_client`` lazily creates an - # ``httpx.AsyncClient`` when no caller client was provided AND a - # ``header_provider`` was set. We treat any client allocated this - # way as owned (closed by the handler). When the caller supplies - # one, we never close it. - owned_client: httpx.AsyncClient | None = None - if provided_client is None: - owned_client = cast("httpx.AsyncClient | None", getattr(tool, "_httpx_client", None)) - return _CacheEntry(tool=tool, owned_httpx_client=owned_client) - - async def _close_invocation_entry(self, entry: _CacheEntry) -> None: - """Finish invocation cleanup even if the caller is cancelled again.""" - # MCPStreamableHTTPTool dispatches connect/close to its lifecycle owner, - # keeping the SDK's cancel-scope entry and exit on that same task. - cleanup = asyncio.create_task(self._close_entry(entry)) + @staticmethod + async def _close_tool(tool: MCPStreamableHTTPTool) -> None: + # The tool owns its SDK lifecycle task and any internally allocated HTTP client. + cleanup = asyncio.create_task(tool.close()) cancelled = False - while not cleanup.done(): - try: - await asyncio.shield(cleanup) - except asyncio.CancelledError: - cancelled = True - cleanup.result() # Propagate cancellation/errors from cleanup itself. - if cancelled: - raise asyncio.CancelledError - - async def _close_entry(self, entry: _CacheEntry) -> None: - """Close the MCP tool and any owned httpx client.""" try: - try: - await entry.tool.close() - except Exception: # pragma: no cover - best effort - logger.debug("DefaultMCPToolHandler: error closing MCP tool", exc_info=True) - finally: - if entry.owned_httpx_client is not None and not entry.owned_httpx_client.is_closed: + while not cleanup.done(): try: - await entry.owned_httpx_client.aclose() - except Exception: # pragma: no cover - best effort - logger.debug("DefaultMCPToolHandler: error closing owned httpx client", exc_info=True) - - @staticmethod - def _cache_key( - server_url: str, - server_label: str | None, - connection_name: str | None, - headers: dict[str, str] | None, - ) -> tuple[str, str, str, str]: - """Build an order-independent cache key for the invocation identity. - - Used only without a ``client_provider``. The key includes - ``server_label`` and ``connection_name`` to distinguish logical - connections. - - Header *names* are lower-cased inside the hash payload only so - that ``Authorization`` and ``authorization`` map to the same - cache entry. Header values remain case-sensitive (per RFC 7235). - """ - if not headers: - headers_hash = "0" - else: - normalized = sorted((k.lower(), v) for k, v in headers.items()) - payload = json.dumps(normalized, ensure_ascii=False) - headers_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest() - return (server_url, server_label or "", connection_name or "", headers_hash) + await asyncio.shield(cleanup) + except asyncio.CancelledError: + cancelled = True + cleanup.result() + except Exception: + logger.warning("DefaultMCPToolHandler: error closing invocation MCP client", exc_info=True) + if cancelled: + raise asyncio.CancelledError diff --git a/python/packages/declarative/tests/conftest.py b/python/packages/declarative/tests/conftest.py index 083de75660f..f1945d22385 100644 --- a/python/packages/declarative/tests/conftest.py +++ b/python/packages/declarative/tests/conftest.py @@ -2,10 +2,19 @@ """Pytest configuration for declarative tests.""" +from __future__ import annotations + import sys +from collections.abc import Callable +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock import pytest +if TYPE_CHECKING: + import httpx2 as httpx + from mcp.server import Server + # Skip all tests in this directory on Python 3.14+ because powerfx doesn't support it yet if sys.version_info >= (3, 14): collect_ignore_glob = ["test_*.py"] @@ -18,3 +27,79 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item for item in items: if "declarative" in str(item.fspath): item.add_marker(skip_marker) + + +@pytest.fixture +def mcp_tool_factory() -> Callable[[], MagicMock]: + from agent_framework import Content, MCPStreamableHTTPTool + from mcp import types + + def create() -> MagicMock: + tool = MagicMock(spec=MCPStreamableHTTPTool) + tool.connect = AsyncMock() + tool.close = AsyncMock() + tool.call_tool = AsyncMock(return_value=[Content.from_text("ok")]) + tool._require_connection.return_value.list_tools_page = AsyncMock(return_value=types.ListToolsResult(tools=[])) + return tool + + return create + + +@pytest.fixture +def mcp_tool(mcp_tool_factory: Callable[[], MagicMock]) -> MagicMock: + return mcp_tool_factory() + + +@pytest.fixture +def mcp_tool_constructor(mcp_tool: MagicMock, monkeypatch: pytest.MonkeyPatch) -> MagicMock: + import agent_framework + + constructor = MagicMock(return_value=mcp_tool) + monkeypatch.setattr(agent_framework, "MCPStreamableHTTPTool", constructor) + return constructor + + +@pytest.fixture +def stateful_mcp_server( + monkeypatch: pytest.MonkeyPatch, +) -> tuple[Server[Any], dict[str, list[str]], list[httpx.AsyncClient]]: + import httpx2 as httpx + from mcp import types + from mcp.server import Server, ServerRequestContext + from mcp.shared.exceptions import MCPError + from starlette.requests import Request + + histories: dict[str, list[str]] = {} + clients: list[httpx.AsyncClient] = [] + client_type = httpx.AsyncClient + + async def reject_discovery(ctx: ServerRequestContext, params: types.RequestParams) -> types.DiscoverResult: + # Force the stateful handshake, not the modern stateless HTTP protocol. + raise MCPError(-32601, "Method not found") + + async def list_tools( + ctx: ServerRequestContext, params: types.PaginatedRequestParams | None + ) -> types.ListToolsResult: + assert isinstance(ctx.request, Request) + histories.setdefault(ctx.request.headers["mcp-session-id"], []) + return types.ListToolsResult( + tools=[types.Tool(name="remember", input_schema={"type": "object", "properties": {}})] + ) + + async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult: + assert isinstance(ctx.request, Request) + history = histories[ctx.request.headers["mcp-session-id"]] + history.append("marker") + return types.CallToolResult(content=[types.TextContent(type="text", text=str(len(history)))]) + + server: Server[Any] = Server("invocation-isolation", on_list_tools=list_tools, on_call_tool=call_tool) + server.add_request_handler("server/discover", types.RequestParams, reject_discovery) + app = server.streamable_http_app(json_response=True) + + def create_client(**kwargs: Any) -> httpx.AsyncClient: + client = client_type(transport=httpx.ASGITransport(app=app), **kwargs) + clients.append(client) + return client + + monkeypatch.setattr(httpx, "AsyncClient", create_client) + return server, histories, clients diff --git a/python/packages/declarative/tests/declarative/test_default_mcp_tool_handler.py b/python/packages/declarative/tests/declarative/test_default_mcp_tool_handler.py new file mode 100644 index 00000000000..8243fb6de1b --- /dev/null +++ b/python/packages/declarative/tests/declarative/test_default_mcp_tool_handler.py @@ -0,0 +1,380 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Test invocation ownership, protocol isolation, and MCP result compatibility.""" + +from __future__ import annotations + +import asyncio +import json +import sys +from collections.abc import Callable +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import httpx2 as httpx +import pytest +from agent_framework import Content, InMemoryCheckpointStorage +from agent_framework.exceptions import ToolException, ToolExecutionException +from mcp import types +from mcp.server import Server +from mcp.shared.exceptions import MCPError + +from agent_framework_declarative._workflows import ToolApprovalResponse, WorkflowFactory +from agent_framework_declarative._workflows._mcp_handler import DefaultMCPToolHandler, MCPToolInvocation + +pytestmark = pytest.mark.skipif(sys.version_info >= (3, 14), reason="PowerFx does not support Python 3.14+") + + +@pytest.mark.parametrize("concurrent", [False, True]) +@pytest.mark.parametrize("provider_mode", ["none", "fallback", "borrowed"]) +async def test_invocations_own_separate_clients( + mcp_tool_factory: Callable[[], MagicMock], + mcp_tool_constructor: MagicMock, + concurrent: bool, + provider_mode: str, +) -> None: + tools = [mcp_tool_factory(), mcp_tool_factory()] + mcp_tool_constructor.side_effect = tools + async with httpx.AsyncClient() as client: + provider = AsyncMock(return_value=client if provider_mode == "borrowed" else None) + handler = DefaultMCPToolHandler(client_provider=None if provider_mode == "none" else provider) + invocation = MCPToolInvocation("https://mcp.example/api", "remember", headers={"X-Test": "same-config"}) + if concurrent: + results = await asyncio.gather(handler.invoke_tool(invocation), handler.invoke_tool(invocation)) + else: + results = (await handler.invoke_tool(invocation), await handler.invoke_tool(invocation)) + assert all(not result.is_error for result in results) + assert mcp_tool_constructor.call_count == 2 + assert provider.await_count == (0 if provider_mode == "none" else 2) + assert not client.is_closed + for tool in tools: + tool.connect.assert_awaited_once() + tool.close.assert_awaited_once() + + +@pytest.mark.parametrize("provider_mode", ["none", "fallback", "borrowed"]) +@pytest.mark.parametrize("concurrent", [False, True]) +async def test_real_protocol_sessions_are_terminated_before_return( + stateful_mcp_server: tuple[Server[Any], dict[str, list[str]], list[httpx.AsyncClient]], + provider_mode: str, + concurrent: bool, +) -> None: + server, histories, clients = stateful_mcp_server + caller_client = httpx.AsyncClient() if provider_mode == "borrowed" else None + provider = AsyncMock(return_value=caller_client) + handler = DefaultMCPToolHandler(client_provider=None if provider_mode == "none" else provider) + invocation = MCPToolInvocation("http://127.0.0.1:8000/mcp", "remember", headers={"X-Test": "same-config"}) + try: + async with server.session_manager.run(): + if concurrent: + results = await asyncio.gather(handler.invoke_tool(invocation), handler.invoke_tool(invocation)) + else: + results = (await handler.invoke_tool(invocation), await handler.invoke_tool(invocation)) + assert [result.outputs[0].text for result in results] == ["1", "1"] + assert len(histories) == 2 + assert not server.session_manager._server_instances + assert all(client.is_closed for client in clients if client is not caller_client) + if caller_client is not None: + assert not caller_client.is_closed + assert not caller_client.event_hooks["request"] + result = await handler.invoke_tool( + MCPToolInvocation(invocation.server_url, "tools/list", headers=dict(invocation.headers)) + ) + assert not result.is_error + assert len(histories) == 3 + assert not server.session_manager._server_instances + finally: + if caller_client is not None: + await caller_client.aclose() + assert all(client.is_closed for client in clients) + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_shared_handler_isolates_independent_workflows( + stateful_mcp_server: tuple[Server[Any], dict[str, list[str]], list[httpx.AsyncClient]], streaming: bool +) -> None: + server, histories, clients = stateful_mcp_server + factory = WorkflowFactory(mcp_tool_handler=DefaultMCPToolHandler()) + definition = { + "name": "isolation", + "actions": [ + { + "kind": "InvokeMcpTool", + "id": "remember", + "serverUrl": "http://127.0.0.1:8000/mcp", + "toolName": "remember", + "headers": {"X-Test": "same-config"}, + } + ], + } + workflows = [factory.create_workflow_from_definition(definition) for _ in range(2)] + async with server.session_manager.run(): + if streaming: + await asyncio.gather(*(workflow.run({}, stream=True).get_final_response() for workflow in workflows)) + else: + await asyncio.gather(*(workflow.run({}) for workflow in workflows)) + assert sorted(map(len, histories.values())) == [1, 1] + assert not server.session_manager._server_instances + assert all(client.is_closed for client in clients) + + +async def test_approval_checkpoint_resume_uses_fresh_protocol_session( + stateful_mcp_server: tuple[Server[Any], dict[str, list[str]], list[httpx.AsyncClient]], +) -> None: + server, histories, clients = stateful_mcp_server + factory = WorkflowFactory(mcp_tool_handler=DefaultMCPToolHandler()) + definition = { + "name": "approval-resume", + "actions": [ + { + "kind": "InvokeMcpTool", + "id": action_id, + "serverUrl": "http://127.0.0.1:8000/mcp", + "toolName": "remember", + "requireApproval": approval, + } + for action_id, approval in (("before_pause", False), ("after_pause", True)) + ], + } + storage = InMemoryCheckpointStorage() + async with server.session_manager.run(): + workflow = factory.create_workflow_from_definition(definition) + await workflow.run({}, checkpoint_storage=storage) + assert sorted(map(len, histories.values())) == [1] + assert not server.session_manager._server_instances + [request] = (await workflow._runner_context.get_pending_request_info_events()).values() + checkpoint_id = workflow.get_last_checkpoint_id() + assert checkpoint_id is not None + resumed = factory.create_workflow_from_definition(definition) + await resumed.run( + checkpoint_id=checkpoint_id, + checkpoint_storage=storage, + responses={request.request_id: ToolApprovalResponse(approved=True)}, + ) + assert sorted(map(len, histories.values())) == [1, 1] + assert not server.session_manager._server_instances + assert all(client.is_closed for client in clients) + + +@pytest.mark.parametrize( + "error", + [ + ToolException("connection failed"), + ToolExecutionException("tool error"), + httpx.ReadTimeout("timeout"), + MCPError(-32603, "MCP error"), + ValueError("invalid configuration"), + ], +) +async def test_connection_errors_clean_up_and_propagate( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, error: Exception +) -> None: + mcp_tool.connect.side_effect = error + with pytest.raises(type(error)) as excinfo: + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + assert excinfo.value is error + mcp_tool.call_tool.assert_not_awaited() + mcp_tool.close.assert_awaited_once() + + +async def test_remote_tool_error_returns_error_result(mcp_tool: MagicMock, mcp_tool_constructor: MagicMock) -> None: + mcp_tool.call_tool.side_effect = ToolExecutionException("tool error") + result = await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + assert result.is_error + assert result.error_message == "tool error" + assert result.outputs[0].text is not None + assert result.outputs[0].text.startswith("Error:") + mcp_tool.close.assert_awaited_once() + + +@pytest.mark.parametrize("wrapped", [False, True]) +@pytest.mark.parametrize("error", [httpx.ReadTimeout("timeout"), MCPError(-32603, "MCP error")]) +async def test_call_transport_errors_clean_up_and_propagate( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, error: Exception, wrapped: bool +) -> None: + failure = ToolExecutionException("Failed to call tool", inner_exception=error) if wrapped else error + if wrapped: + failure.__cause__ = error + mcp_tool.call_tool.side_effect = failure + with pytest.raises(type(failure)) as excinfo: + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + assert excinfo.value is failure + if wrapped: + assert excinfo.value.__cause__ is error + mcp_tool.close.assert_awaited_once() + + +@pytest.mark.parametrize("stage", ["connect", "call_tool"]) +async def test_cancellation_cleans_up_and_propagates( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, stage: str +) -> None: + getattr(mcp_tool, stage).side_effect = asyncio.CancelledError + with pytest.raises(asyncio.CancelledError): + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + mcp_tool.close.assert_awaited_once() + + +async def test_repeated_cancellation_waits_for_cleanup(mcp_tool: MagicMock, mcp_tool_constructor: MagicMock) -> None: + release = asyncio.Event() + mcp_tool.close.side_effect = release.wait + task = asyncio.create_task( + DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + ) + try: + for _ in range(10): + await asyncio.sleep(0) + if mcp_tool.close.await_count: + break + assert mcp_tool.close.await_count == 1 + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_unexpected_call_error_propagates(mcp_tool: MagicMock, mcp_tool_constructor: MagicMock) -> None: + mcp_tool.call_tool.side_effect = RuntimeError("programming error") + with pytest.raises(RuntimeError, match="programming error"): + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + mcp_tool.close.assert_awaited_once() + + +@pytest.mark.parametrize("call_fails", [False, True]) +async def test_cleanup_error_does_not_replace_invocation_outcome( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, caplog: pytest.LogCaptureFixture, call_fails: bool +) -> None: + mcp_tool.close.side_effect = RuntimeError("cleanup failed") + handler = DefaultMCPToolHandler() + invocation = MCPToolInvocation("https://mcp.example/api", "remember") + if call_fails: + mcp_tool.call_tool.side_effect = RuntimeError("call failed") + with pytest.raises(RuntimeError, match="call failed"): + await handler.invoke_tool(invocation) + else: + result = await handler.invoke_tool(invocation) + assert not result.is_error + assert "error closing invocation MCP client" in caplog.text + + +@pytest.mark.parametrize("error", [RuntimeError("provider failed"), httpx.ConnectError("connection failed")]) +async def test_provider_failure_propagates_without_constructing_tool( + mcp_tool_constructor: MagicMock, error: Exception +) -> None: + provider = AsyncMock(side_effect=error) + with pytest.raises(type(error)) as excinfo: + await DefaultMCPToolHandler(client_provider=provider).invoke_tool( + MCPToolInvocation("https://mcp.example/api", "remember") + ) + assert excinfo.value is error + mcp_tool_constructor.assert_not_called() + + +@pytest.mark.parametrize("raw", ["plain text", [Content.from_text("first"), Content.from_text("second")]]) +async def test_result_normalisation(mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, raw: Any) -> None: + mcp_tool.call_tool.return_value = raw + result = await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "remember")) + assert [content.text for content in result.outputs] == ( + ["plain text"] if isinstance(raw, str) else ["first", "second"] + ) + + +async def test_constructor_passes_invocation_and_discovery_options(mcp_tool_constructor: MagicMock) -> None: + invocation = MCPToolInvocation( + "https://mcp.example/api", "remember", server_label="label", headers={"X-Test": "value"} + ) + await DefaultMCPToolHandler(max_discovery_pages=12, max_discovery_items=345).invoke_tool(invocation) + assert mcp_tool_constructor.call_args.kwargs == { + "name": "label", + "url": invocation.server_url, + "load_prompts": False, + "http_client": None, + "static_headers": {"X-Test": "value"}, + "max_discovery_pages": 12, + "max_discovery_items": 345, + } + assert mcp_tool_constructor.call_args.kwargs["static_headers"] is not invocation.headers + + +@pytest.mark.parametrize("option", ["max_discovery_pages", "max_discovery_items"]) +@pytest.mark.parametrize("value", [0, -1]) +def test_invalid_discovery_limit_raises(option: str, value: int) -> None: + with pytest.raises(ValueError, match=f"{option} must be positive or None"): + if option == "max_discovery_pages": + DefaultMCPToolHandler(max_discovery_pages=value) + else: + DefaultMCPToolHandler(max_discovery_items=value) + + +async def test_list_tools_rejects_arguments_before_provider(mcp_tool_constructor: MagicMock) -> None: + provider = AsyncMock() + with pytest.raises(ValueError, match="The reserved MCP 'tools/list' operation does not accept tool arguments"): + await DefaultMCPToolHandler(client_provider=provider).invoke_tool( + MCPToolInvocation("https://mcp.example/api", "tools/list", arguments={"unexpected": True}) + ) + provider.assert_not_awaited() + mcp_tool_constructor.assert_not_called() + + +async def test_list_tools_preserves_catalog_and_pagination( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock +) -> None: + list_page = mcp_tool._require_connection.return_value.list_tools_page + list_page.side_effect = [ + types.ListToolsResult( + tools=[ + types.Tool( + name="first", + description="description", + input_schema={"type": "object"}, + output_schema={"type": "string"}, + ) + ], + next_cursor="next", + ), + types.ListToolsResult(tools=[types.Tool(name="second", input_schema={"type": "object"})]), + ] + result = await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "tools/list")) + text = result.outputs[0].text + assert text is not None + assert json.loads(text) == { + "tools": [ + { + "name": "first", + "description": "description", + "inputSchema": {"type": "object"}, + "outputSchema": {"type": "string"}, + }, + {"name": "second", "description": None, "inputSchema": {"type": "object"}, "outputSchema": None}, + ] + } + assert list(json.loads(text)["tools"][0]) == ["name", "description", "inputSchema", "outputSchema"] + assert "\n " in text + assert list_page.await_args_list[0].args == (None,) + assert list_page.await_args_list[1].args[0].cursor == "next" + mcp_tool.call_tool.assert_not_awaited() + mcp_tool.close.assert_awaited_once() + + +async def test_list_tools_disconnected_error(mcp_tool: MagicMock, mcp_tool_constructor: MagicMock) -> None: + mcp_tool._require_connection.side_effect = RuntimeError("disconnected") + with pytest.raises(RuntimeError, match="disconnected"): + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "tools/list")) + mcp_tool.close.assert_awaited_once() + + +@pytest.mark.parametrize( + "error", [httpx.ReadTimeout("timeout"), MCPError(-32603, "MCP error"), ToolExecutionException("discovery failed")] +) +async def test_list_tools_transport_error( + mcp_tool: MagicMock, mcp_tool_constructor: MagicMock, error: Exception +) -> None: + mcp_tool._require_connection.return_value.list_tools_page.side_effect = error + with pytest.raises(type(error)) as excinfo: + await DefaultMCPToolHandler().invoke_tool(MCPToolInvocation("https://mcp.example/api", "tools/list")) + assert excinfo.value is error + mcp_tool.close.assert_awaited_once() diff --git a/python/packages/declarative/tests/test_default_mcp_tool_handler.py b/python/packages/declarative/tests/test_default_mcp_tool_handler.py deleted file mode 100644 index 03a1f880465..00000000000 --- a/python/packages/declarative/tests/test_default_mcp_tool_handler.py +++ /dev/null @@ -1,1475 +0,0 @@ -# Copyright (c) Microsoft. All rights reserved. - -"""Tests for ``DefaultMCPToolHandler``. - -Most tests exercise the real handler against a fake ``MCPStreamableHTTPTool`` -(no real MCP server, no real network) to cover the parts of the handler not -exercisable through the executor stub: cache hit/miss/eviction, concurrent -connect via in-flight futures, header isolation across cache keys, -string-result normalisation, ``load_prompts=False`` verification, and -owned-vs-caller httpx close semantics, and provider-backed invocation lifetimes. - -Two lifecycle cases use the real MCP SDK with an inert HTTPX mock transport. -""" - -from __future__ import annotations - -import asyncio -import json -import sys -from typing import Any -from unittest.mock import AsyncMock, patch - -import httpx2 as httpx -import pytest -from agent_framework import Content -from agent_framework.exceptions import ToolExecutionException - -from agent_framework_declarative._workflows._mcp_handler import ( - DefaultMCPToolHandler, - MCPToolInvocation, -) - -pytestmark = pytest.mark.skipif( - sys.version_info >= (3, 14), - reason="Skipped on Python 3.14+ to keep parity with rest of declarative suite", -) - - -class FakeListToolsResult: # noqa: B903 - mimics ``mcp.types.ListToolsResult`` shape, not a value type - """Stand-in for ``mcp.types.ListToolsResult`` returned by ``session.list_tools()``.""" - - def __init__(self, tools: list[Any], next_cursor: str | None = None) -> None: - self.tools = tools - self.nextCursor = next_cursor - - -class FakeMcpTool: - """Stand-in for an MCP ``Tool`` (subset used by ``_invoke_list_tools``).""" - - def __init__( - self, - name: str, - description: str | None = None, - inputSchema: dict[str, Any] | None = None, - outputSchema: dict[str, Any] | None = None, - ) -> None: - self.name = name - self.description = description - self.inputSchema = inputSchema if inputSchema is not None else {"type": "object", "properties": {}} - self.outputSchema = outputSchema - - -class FakeMcpSession: - """Stand-in for ``mcp.ClientSession``. - - ``list_tools_pages`` lets a test enqueue multiple paginated responses; - when None (default), an empty single-page result is returned. ``list_tools_error`` - raises a synthetic error on the next call when set. - """ - - def __init__(self) -> None: - self.list_tools_pages: list[FakeListToolsResult] | None = None - self.list_tools_calls: list[Any] = [] - self.list_tools_error: BaseException | None = None - - async def list_tools(self, params: Any = None) -> FakeListToolsResult: - self.list_tools_calls.append(params) - if self.list_tools_error is not None: - raise self.list_tools_error - if self.list_tools_pages is None: - return FakeListToolsResult(tools=[]) - index = len(self.list_tools_calls) - 1 - if index >= len(self.list_tools_pages): - return FakeListToolsResult(tools=[]) - return self.list_tools_pages[index] - - -class FakeTool: - """Stand-in for ``MCPStreamableHTTPTool``. - - Records constructor kwargs, tracks connect/close lifecycle, and dispatches - ``call_tool`` to a per-instance handler. - """ - - instances: list[FakeTool] = [] - - def __init__(self, **kwargs: Any) -> None: - self.kwargs = kwargs - self.connect_count = 0 - self.close_count = 0 - self.connect_delay: float = 0.0 - self.connect_error: BaseException | None = None - self.call_handler: Any = lambda **_a: [Content.from_text("ok")] - self._httpx_client: httpx.AsyncClient | None = kwargs.get("http_client") - self.session: FakeMcpSession | None = None - # Mimic MCPStreamableHTTPTool: when no caller client AND header_provider - # is set, lazily allocate an owned httpx client during connect. - FakeTool.instances.append(self) - - async def connect(self) -> None: - if self.connect_delay: - await asyncio.sleep(self.connect_delay) - if self.connect_error is not None: - raise self.connect_error - self.connect_count += 1 - # Mimic lazy httpx allocation when no client provided AND header_provider set. - if self.kwargs.get("http_client") is None and self.kwargs.get("header_provider") is not None: - self._httpx_client = httpx.AsyncClient() - # Mimic MCPStreamableHTTPTool: a live session becomes available after connect. - if self.session is None: - self.session = FakeMcpSession() - - async def close(self) -> None: - self.close_count += 1 - - async def call_tool(self, tool_name: str, **arguments: Any) -> Any: - return self.call_handler(tool_name=tool_name, **arguments) - - -@pytest.fixture(autouse=True) -def _clear_fake_instances() -> None: - FakeTool.instances.clear() - - -def _patch_tool() -> Any: - """Patch the lazy import inside ``_create_entry`` to substitute FakeTool.""" - import agent_framework - - return patch.object(agent_framework, "MCPStreamableHTTPTool", FakeTool) - - -def _invocation( - *, server_url: str = "https://mcp.example/api", tool_name: str = "search", **overrides: Any -) -> MCPToolInvocation: - return MCPToolInvocation( - server_url=server_url, - tool_name=tool_name, - **overrides, - ) - - -# ---------- Provider-backed invocation lifetimes -------------------------- - - -class TestProviderLifetimes: - @pytest.mark.parametrize("returns_none", [False, True]) - async def test_actual_sdk_lifecycle_and_hook_cleanup( - self, returns_none: bool, caplog: pytest.LogCaptureFixture - ) -> None: - import agent_framework - from agent_framework import MCPStreamableHTTPTool - - caplog.set_level("DEBUG", logger="agent_framework_declarative._workflows._mcp_handler") - initialized: list[str] = [] - terminated: list[str] = [] - methods: list[str] = [] - tools: list[MCPStreamableHTTPTool] = [] - clients: list[RecordingClient] = [] - caller_hook = AsyncMock() - - async def respond(request: httpx.Request) -> httpx.Response: - assert request.headers["X-Test"] == "lifecycle" - if request.method == "GET": - return httpx.Response(405) - if request.method == "DELETE": - session_id = request.headers.get("mcp-session-id") - if session_id is not None: - terminated.append(session_id) - return httpx.Response(200) - - body = json.loads(request.content) - method = body["method"] - methods.append(method) - if "id" not in body: - return httpx.Response(202) - headers: dict[str, str] = {} - result: dict[str, Any] - if method == "server/discover": - initialized.append(f"discovery-{len(initialized) + 1}") - result = { - "supportedVersions": ["2026-07-28"], - "capabilities": {"tools": {}}, - } - elif method == "tools/list": - result = { - "resultType": "complete", - "ttlMs": 0, - "cacheScope": "private", - "tools": [{"name": "ok", "inputSchema": {"type": "object", "properties": {}}}], - } - elif method == "tools/call": - result = {"resultType": "complete", "content": [{"type": "text", "text": "ok"}]} - else: - assert method == "ping" - result = {} - return httpx.Response(200, headers=headers, json={"jsonrpc": "2.0", "id": body["id"], "result": result}) - - class RecordingClient(httpx.AsyncClient): - def __init__(self, **kwargs: Any) -> None: - super().__init__( - transport=httpx.MockTransport(respond), event_hooks={"request": [caller_hook]}, **kwargs - ) - self.close_count = 0 - clients.append(self) - - async def aclose(self) -> None: - self.close_count += 1 - await super().aclose() - - def create_tool(**kwargs: Any) -> MCPStreamableHTTPTool: - tool = MCPStreamableHTTPTool(**kwargs) - tools.append(tool) - return tool - - caller_client = None if returns_none else RecordingClient() - provider = AsyncMock(return_value=caller_client) - try: - with ( - patch.object(httpx, "AsyncClient", RecordingClient), - patch.object(agent_framework, "MCPStreamableHTTPTool", side_effect=create_tool), - ): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - for index, tool_name in enumerate(("ok", "tools/list", "ok", "tools/list"), start=1): - result = await handler.invoke_tool( - _invocation(tool_name=tool_name, headers={"X-Test": "lifecycle"}) - ) - assert not result.is_error - if tool_name == "ok": - assert result.outputs[0].text == "ok" - else: - assert result.outputs[0].text is not None - assert json.loads(result.outputs[0].text)["tools"][0]["name"] == "ok" - assert provider.await_count == index - assert len(tools) == index - assert len(initialized) == index - assert not terminated - assert not handler._active_invocations - assert not handler._cache - tool = tools[-1] - assert tool.session is None - assert not tool.is_connected - assert tool._lifecycle_owner_task is None - assert tool._header_hook_client is None - assert all(client.event_hooks["request"] == [caller_hook] for client in clients) - if returns_none: - assert len(clients) == index - assert all(client.is_closed and client.close_count == 1 for client in clients) - else: - assert caller_client is not None - assert not caller_client.is_closed - assert caller_client.close_count == 0 - assert len({id(tool) for tool in tools}) == 4 - assert methods.count("server/discover") == 4 - assert methods.count("tools/call") == 2 - assert methods.count("tools/list") == 6 - assert not any( - "Could not cleanly close MCP exit stack" in record.getMessage() - or "DefaultMCPToolHandler: error closing" in record.getMessage() - for record in caplog.records - ) - finally: - for client in clients: - if not client.is_closed: - await client.aclose() - - @pytest.mark.parametrize("returns_none", [False, True]) - async def test_sequential_invocations_get_fresh_tools(self, returns_none: bool) -> None: - async with httpx.AsyncClient() as client: - provider = AsyncMock(return_value=None if returns_none else client) - with _patch_tool(): - async with DefaultMCPToolHandler(client_provider=provider, cache_max_size=1) as handler: - for index, tool_name in enumerate(("search", "tools/list", "search", "tools/list"), start=1): - invocation = _invocation(tool_name=tool_name, headers={"X-Test": "1"}) - result = await handler.invoke_tool(invocation) - assert not result.is_error - assert result.outputs[0].text == ('{\n "tools": []\n}' if tool_name == "tools/list" else "ok") - assert provider.await_count == index - assert provider.await_args is not None - assert provider.await_args.args[0] is invocation - assert len(FakeTool.instances) == index - tool = FakeTool.instances[-1] - assert tool.connect_count == 1 - assert tool.close_count == 1 - assert tool.kwargs["http_client"] is (None if returns_none else client) - if returns_none: - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - assert not client.is_closed - assert not handler._cache - assert not handler._inflight - assert not handler._active_invocations - assert len({id(tool.session) for tool in FakeTool.instances}) == 4 - assert all(tool.close_count == 1 for tool in FakeTool.instances) - assert not client.is_closed - - @pytest.mark.parametrize("returns_none", [False, True]) - async def test_concurrent_invocations_get_fresh_tools(self, returns_none: bool) -> None: - all_connecting = asyncio.Event() - release = asyncio.Event() - original_connect = FakeTool.connect - - async def gated_connect(tool: FakeTool) -> None: - await original_connect(tool) - if len(FakeTool.instances) == 3: - all_connecting.set() - await release.wait() - - async with httpx.AsyncClient() as client: - provider = AsyncMock(return_value=None if returns_none else client) - with _patch_tool(), patch.object(FakeTool, "connect", gated_connect): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - tasks = [ - asyncio.create_task(handler.invoke_tool(_invocation(tool_name=name, headers={"X-Test": "1"}))) - for name in ("search", "tools/list", "search") - ] - try: - await asyncio.wait_for(all_connecting.wait(), timeout=5) - assert provider.await_count == 3 - assert len({id(tool.session) for tool in FakeTool.instances}) == 3 - assert all(tool.close_count == 0 for tool in FakeTool.instances) - assert not handler._cache - assert not handler._inflight - finally: - release.set() - results = await asyncio.gather(*tasks) - assert all(not result.is_error for result in results) - assert all(tool.close_count == 1 for tool in FakeTool.instances) - if returns_none: - assert all(tool._httpx_client and tool._httpx_client.is_closed for tool in FakeTool.instances) - assert not handler._active_invocations - assert not client.is_closed - - @pytest.mark.timeout(10) - @pytest.mark.parametrize("stage", ["provider", "connect", "search", "tools/list", "close"]) - @pytest.mark.parametrize("in_child_task", [False, True]) - async def test_reentrant_shutdown_is_rejected_without_closing_handler( - self, stage: str, in_child_task: bool - ) -> None: - original_connect = FakeTool.connect - original_close = FakeTool.close - original_call = FakeTool.call_tool - original_list = FakeMcpSession.list_tools - rejected = 0 - - async def check_shutdown(phase: str) -> None: - nonlocal rejected - if stage != phase: - return - with pytest.raises(RuntimeError, match="cannot be called from an active provider-backed invocation"): - if in_child_task: - await asyncio.create_task(handler.aclose()) - else: - await handler.aclose() - rejected += 1 - assert not handler._closed - - async def provider(_invocation: MCPToolInvocation) -> None: - await check_shutdown("provider") - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - await check_shutdown("connect") - - async def close(tool: FakeTool) -> None: - await check_shutdown("close") - await original_close(tool) - - async def call(tool: FakeTool, tool_name: str, **arguments: Any) -> Any: - await check_shutdown("search") - return await original_call(tool, tool_name, **arguments) - - async def list_tools(session: FakeMcpSession, params: Any = None) -> FakeListToolsResult: - await check_shutdown("tools/list") - return await original_list(session, params) - - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "close", close), - patch.object(FakeTool, "call_tool", call), - patch.object(FakeMcpSession, "list_tools", list_tools), - ): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - for _ in range(2): - result = await handler.invoke_tool( - _invocation(tool_name="tools/list" if stage == "tools/list" else "search") - ) - assert not result.is_error - assert handler._invocation_context.get() == () - assert not handler._active_invocations - assert not handler._closed - assert rejected == 2 - assert len(FakeTool.instances) == 2 - assert all(tool.close_count == 1 for tool in FakeTool.instances) - - @pytest.mark.timeout(10) - async def test_nested_invocations_restore_context_and_keep_active_ancestry(self) -> None: - release_child = asyncio.Event() - children: list[asyncio.Task[None]] = [] - - async def close_from_inherited_context() -> None: - await release_child.wait() - outer, inner = handler._invocation_context.get() - assert not outer.done() - assert inner.done() - with pytest.raises(RuntimeError, match="cannot be called from an active provider-backed invocation"): - await handler.aclose() - assert not handler._closed - - async def provider(invocation: MCPToolInvocation) -> None: - if invocation.tool_name == "inner": - children.append(asyncio.create_task(close_from_inherited_context())) - return - outer_context = handler._invocation_context.get() - assert len(outer_context) == 1 - nested_result = await handler.invoke_tool(_invocation(tool_name="inner")) - assert not nested_result.is_error - assert handler._invocation_context.get() is outer_context - release_child.set() - await children[-1] - with pytest.raises(RuntimeError, match="cannot be called from an active provider-backed invocation"): - await handler.aclose() - - with _patch_tool(): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - try: - result = await handler.invoke_tool(_invocation(tool_name="outer")) - assert not result.is_error - assert handler._invocation_context.get() == () - assert not handler._active_invocations - finally: - release_child.set() - await asyncio.gather(*children) - assert len(FakeTool.instances) == 2 - assert all(tool.close_count == 1 for tool in FakeTool.instances) - - @pytest.mark.timeout(10) - async def test_provider_can_invoke_and_close_another_handler(self) -> None: - async def provider(_invocation: MCPToolInvocation) -> None: - async with DefaultMCPToolHandler(client_provider=AsyncMock(return_value=None)) as other: - result = await other.invoke_tool(_invocation) - assert not result.is_error - assert other._invocation_context.get() == () - assert handler._invocation_context.get() - assert other._closed - assert not handler._closed - - with _patch_tool(): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - result = await handler.invoke_tool(_invocation()) - assert not result.is_error - assert handler._invocation_context.get() == () - assert len(FakeTool.instances) == 2 - assert all(tool.close_count == 1 for tool in FakeTool.instances) - - @pytest.mark.timeout(10) - async def test_inherited_context_can_close_after_invocation_completes(self) -> None: - release_child = asyncio.Event() - children: list[asyncio.Task[None]] = [] - - async def close_from_inherited_context() -> None: - await release_child.wait() - context = handler._invocation_context.get() - assert len(context) == 1 - assert context[0].done() - await handler.aclose() - - async def provider(_invocation: MCPToolInvocation) -> None: - children.append(asyncio.create_task(close_from_inherited_context())) - - with _patch_tool(): - handler = DefaultMCPToolHandler(client_provider=provider) - try: - result = await handler.invoke_tool(_invocation()) - assert not result.is_error - assert handler._invocation_context.get() == () - assert not handler._active_invocations - assert not handler._closed - finally: - release_child.set() - await asyncio.gather(*children) - await handler.aclose() - assert handler._closed - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].close_count == 1 - - async def test_list_tools_paginates_within_each_invocation(self) -> None: - original_connect = FakeTool.connect - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - assert tool.session is not None - tool.session.list_tools_pages = [ - FakeListToolsResult([FakeMcpTool("first")], next_cursor="next"), - FakeListToolsResult([FakeMcpTool("second")]), - ] - - provider = AsyncMock(return_value=None) - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "call_tool", new_callable=AsyncMock) as call_tool, - ): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - for _ in range(2): - result = await handler.invoke_tool(_invocation(tool_name="tools/list")) - assert not result.is_error - assert result.outputs[0].text is not None - payload = json.loads(result.outputs[0].text) - assert [tool["name"] for tool in payload["tools"]] == ["first", "second"] - call_tool.assert_not_awaited() - assert provider.await_count == 2 - assert len(FakeTool.instances) == 2 - for tool in FakeTool.instances: - assert tool.close_count == 1 - assert tool.session is not None - assert len(tool.session.list_tools_calls) == 2 - assert tool.session.list_tools_calls[0] is None - assert tool.session.list_tools_calls[1].cursor == "next" - - @pytest.mark.parametrize("stage", ["provider", "connect", "search", "tools/list"]) - @pytest.mark.parametrize( - "error_type", [httpx.ReadTimeout, ToolExecutionException, RuntimeError, asyncio.CancelledError] - ) - @pytest.mark.parametrize("returns_none", [False, True]) - async def test_errors_clean_up_and_preserve_mapping( - self, stage: str, error_type: type[BaseException], returns_none: bool - ) -> None: - error = error_type("operation stopped") - original_connect = FakeTool.connect - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - if stage == "connect": - raise error - - async with httpx.AsyncClient() as client: - provider = AsyncMock( - return_value=None if returns_none else client, - side_effect=error if stage == "provider" else None, - ) - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "call_tool", new_callable=AsyncMock, side_effect=error), - patch.object(FakeMcpSession, "list_tools", new_callable=AsyncMock, side_effect=error), - ): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - invocation = _invocation( - tool_name="tools/list" if stage == "tools/list" else "search", headers={"X-Test": "1"} - ) - if error_type is asyncio.CancelledError or ( - error_type is RuntimeError and stage in ("search", "tools/list") - ): - with pytest.raises(error_type, match="operation stopped"): - await handler.invoke_tool(invocation) - else: - result = await handler.invoke_tool(invocation) - assert result.is_error - assert "operation stopped" in (result.error_message or "") - assert result.outputs[0].text is not None - assert result.outputs[0].text.startswith("Error:") - assert handler._invocation_context.get() == () - assert not handler._active_invocations - assert not handler._cache - assert not handler._inflight - assert provider.await_count == 1 - assert len(FakeTool.instances) == (0 if stage == "provider" else 1) - for tool in FakeTool.instances: - assert tool.close_count == 1 - if returns_none: - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - assert not client.is_closed - - @pytest.mark.parametrize("tool_name", ["search", "tools/list"]) - async def test_mcp_error_mapping_cleans_up(self, tool_name: str) -> None: - from mcp.shared.exceptions import MCPError - - error = MCPError(-32603, "operation stopped") - with ( - _patch_tool(), - patch.object(FakeTool, "call_tool", new_callable=AsyncMock, side_effect=error), - patch.object(FakeMcpSession, "list_tools", new_callable=AsyncMock, side_effect=error), - ): - async with DefaultMCPToolHandler(client_provider=AsyncMock(return_value=None)) as handler: - result = await handler.invoke_tool(_invocation(tool_name=tool_name, headers={"X-Test": "1"})) - assert result.is_error - assert result.error_message == "operation stopped" - assert not handler._active_invocations - tool = FakeTool.instances[0] - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - async def test_tool_close_failure_still_closes_fallback_client(self) -> None: - async def close(tool: FakeTool) -> None: - tool.close_count += 1 - raise RuntimeError("close failed") - - with _patch_tool(), patch.object(FakeTool, "close", close): - async with DefaultMCPToolHandler(client_provider=AsyncMock(return_value=None)) as handler: - result = await handler.invoke_tool(_invocation(headers={"X-Test": "1"})) - assert not result.is_error - assert result.outputs[0].text == "ok" - assert not handler._active_invocations - tool = FakeTool.instances[0] - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - @pytest.mark.parametrize("connect_fails", [False, True]) - async def test_cleanup_cancellation_signals_completion(self, connect_fails: bool) -> None: - original_connect = FakeTool.connect - completions: list[asyncio.Future[None]] = [] - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - if connect_fails: - raise httpx.ConnectError("connect stopped") - - async def close(tool: FakeTool) -> None: - tool.close_count += 1 - completions.extend(handler._active_invocations) - raise asyncio.CancelledError("cleanup stopped") - - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "close", close), - ): - handler = DefaultMCPToolHandler(client_provider=AsyncMock(return_value=None)) - with pytest.raises(asyncio.CancelledError, match="cleanup stopped"): - await handler.invoke_tool(_invocation(headers={"X-Test": "1"})) - assert handler._invocation_context.get() == () - assert not handler._active_invocations - assert len(completions) == 1 - assert completions[0].done() - assert completions[0].result() is None - await asyncio.wait_for(handler.aclose(), timeout=5) - tool = FakeTool.instances[0] - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - @pytest.mark.parametrize("stage", ["provider", "connect", "search", "tools/list"]) - async def test_task_cancellation_cleans_up(self, stage: str) -> None: - started = asyncio.Event() - original_connect = FakeTool.connect - - async def wait(*_args: Any, **_kwargs: Any) -> None: - started.set() - await asyncio.Event().wait() - - async def provider(_invocation: MCPToolInvocation) -> None: - if stage == "provider": - await wait() - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - if stage == "connect": - await wait() - - async def invoke_and_check_context() -> None: - try: - await handler.invoke_tool( - _invocation(tool_name="tools/list" if stage == "tools/list" else "search", headers={"X-Test": "1"}) - ) - finally: - assert handler._invocation_context.get() == () - - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "call_tool", new_callable=AsyncMock, side_effect=wait), - patch.object(FakeMcpSession, "list_tools", new_callable=AsyncMock, side_effect=wait), - ): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - task = asyncio.create_task(invoke_and_check_context()) - await asyncio.wait_for(started.wait(), timeout=5) - task.cancel() - with pytest.raises(asyncio.CancelledError): - await task - assert not handler._active_invocations - assert len(FakeTool.instances) == (0 if stage == "provider" else 1) - for tool in FakeTool.instances: - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - async def test_repeated_cancellation_waits_for_cleanup(self) -> None: - cleanup_started = asyncio.Event() - release_cleanup = asyncio.Event() - original_close = FakeTool.close - - async def close(tool: FakeTool) -> None: - cleanup_started.set() - await release_cleanup.wait() - await original_close(tool) - - with _patch_tool(), patch.object(FakeTool, "close", close): - async with DefaultMCPToolHandler(client_provider=AsyncMock(return_value=None)) as handler: - task = asyncio.create_task(handler.invoke_tool(_invocation(headers={"X-Test": "1"}))) - await asyncio.wait_for(cleanup_started.wait(), timeout=5) - for _ in range(2): - task.cancel() - await asyncio.sleep(0) - assert not task.done() - assert handler._active_invocations - release_cleanup.set() - with pytest.raises(asyncio.CancelledError): - await task - assert not handler._active_invocations - tool = FakeTool.instances[0] - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - @pytest.mark.parametrize("stage", ["provider", "connect", "search", "tools/list", "close"]) - @pytest.mark.parametrize("cancel_shutdown", [False, True]) - async def test_shutdown_waits_for_invocation_cleanup(self, stage: str, cancel_shutdown: bool) -> None: - started = asyncio.Event() - release = asyncio.Event() - original_connect = FakeTool.connect - original_close = FakeTool.close - original_call = FakeTool.call_tool - original_list = FakeMcpSession.list_tools - - async def gate(phase: str) -> None: - if stage == phase: - started.set() - await release.wait() - - async def provider(_invocation: MCPToolInvocation) -> None: - await gate("provider") - - async def connect(tool: FakeTool) -> None: - await original_connect(tool) - await gate("connect") - - async def close(tool: FakeTool) -> None: - await gate("close") - await original_close(tool) - - async def call(tool: FakeTool, tool_name: str, **arguments: Any) -> Any: - await gate("search") - return await original_call(tool, tool_name, **arguments) - - async def list_tools(session: FakeMcpSession, params: Any = None) -> FakeListToolsResult: - await gate("tools/list") - return await original_list(session, params) - - provider_mock = AsyncMock(side_effect=provider) - with ( - _patch_tool(), - patch.object(FakeTool, "connect", connect), - patch.object(FakeTool, "close", close), - patch.object(FakeTool, "call_tool", call), - patch.object(FakeMcpSession, "list_tools", list_tools), - ): - handler = DefaultMCPToolHandler(client_provider=provider_mock) - invocation = _invocation( - tool_name="tools/list" if stage == "tools/list" else "search", headers={"X-Test": "1"} - ) - task = asyncio.create_task(handler.invoke_tool(invocation)) - await asyncio.wait_for(started.wait(), timeout=5) - shutdown = asyncio.create_task(handler.aclose()) - await asyncio.sleep(0) - assert handler._closed - assert not shutdown.done() - rejected = await handler.invoke_tool(invocation) - assert rejected.is_error - assert "closed" in (rejected.error_message or "") - assert provider_mock.await_count == 1 - if cancel_shutdown: - shutdown.cancel() - with pytest.raises(asyncio.CancelledError): - await shutdown - second_shutdown = asyncio.create_task(handler.aclose()) - await asyncio.sleep(0) - assert not second_shutdown.done() - release.set() - result = await task - await second_shutdown - if not cancel_shutdown: - await shutdown - await handler.aclose() - assert result.is_error == (stage in ("provider", "connect")) - if result.is_error: - assert "closed" in (result.error_message or "") - assert not handler._active_invocations - assert not handler._cache - assert not handler._inflight - assert len(FakeTool.instances) == (0 if stage == "provider" else 1) - for tool in FakeTool.instances: - assert tool.close_count == 1 - assert tool._httpx_client is not None - assert tool._httpx_client.is_closed - - async def test_invalid_list_arguments_skip_provider(self) -> None: - provider = AsyncMock(return_value=None) - with _patch_tool(): - async with DefaultMCPToolHandler(client_provider=provider) as handler: - result = await handler.invoke_tool(_invocation(tool_name="tools/list", arguments={"q": "test"})) - assert result.is_error - assert "does not accept tool arguments" in (result.error_message or "") - provider.assert_not_awaited() - assert not FakeTool.instances - - -# ---------- Construction --------------------------------------------------- - - -class TestConstruction: - def test_invalid_cache_size_raises(self) -> None: - with pytest.raises(ValueError): - DefaultMCPToolHandler(cache_max_size=0) - with pytest.raises(ValueError): - DefaultMCPToolHandler(cache_max_size=-3) - - def test_invalid_discovery_limit_raises(self) -> None: - with pytest.raises(ValueError, match="max_discovery_pages must be positive or None"): - DefaultMCPToolHandler(max_discovery_pages=0) - with pytest.raises(ValueError, match="max_discovery_items must be positive or None"): - DefaultMCPToolHandler(max_discovery_items=-1) - - -# ---------- Tool kwargs ---------------------------------------------------- - - -class TestToolKwargs: - @pytest.mark.asyncio - async def test_discovery_limits_passed_to_tool(self) -> None: - handler = DefaultMCPToolHandler(max_discovery_pages=12, max_discovery_items=345) - with _patch_tool(): - await handler.invoke_tool(_invocation()) - - assert FakeTool.instances[0].kwargs["max_discovery_pages"] == 12 - assert FakeTool.instances[0].kwargs["max_discovery_items"] == 345 - - @pytest.mark.asyncio - async def test_load_prompts_false_passed_to_tool(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].kwargs["load_prompts"] is False - - @pytest.mark.asyncio - async def test_server_label_used_as_tool_name(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(server_label="MyMcp")) - assert FakeTool.instances[0].kwargs["name"] == "MyMcp" - - @pytest.mark.asyncio - async def test_default_tool_name_when_no_label(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(server_label=None)) - assert FakeTool.instances[0].kwargs["name"] == "McpClient" - - @pytest.mark.asyncio - async def test_no_header_provider_when_no_headers(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={})) - assert FakeTool.instances[0].kwargs["header_provider"] is None - - @pytest.mark.asyncio - async def test_header_provider_returns_captured_headers(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"Authorization": "Bearer T"})) - provider = FakeTool.instances[0].kwargs["header_provider"] - assert provider({}) == {"Authorization": "Bearer T"} - # Even if runtime kwargs change, captured headers stay the same. - assert provider({"foo": "bar"}) == {"Authorization": "Bearer T"} - - -# ---------- Cache behaviour ------------------------------------------------ - - -class TestCache: - @pytest.mark.asyncio - async def test_same_url_and_headers_hit_cache(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"X": "1"})) - await handler.invoke_tool(_invocation(headers={"X": "1"})) - # One tool created, connect called once. - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].connect_count == 1 - - @pytest.mark.asyncio - async def test_different_headers_create_separate_entries(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"Authorization": "tk-A"})) - await handler.invoke_tool(_invocation(headers={"Authorization": "tk-B"})) - assert len(FakeTool.instances) == 2 - - @pytest.mark.asyncio - async def test_different_urls_create_separate_entries(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(server_url="https://mcp.a/api")) - await handler.invoke_tool(_invocation(server_url="https://mcp.b/api")) - assert len(FakeTool.instances) == 2 - - @pytest.mark.asyncio - async def test_lru_eviction_closes_old_entry(self) -> None: - handler = DefaultMCPToolHandler(cache_max_size=2) - with _patch_tool(): - await handler.invoke_tool(_invocation(server_url="https://a/")) - await handler.invoke_tool(_invocation(server_url="https://b/")) - # Inserting a third evicts the LRU entry (the first one). - await handler.invoke_tool(_invocation(server_url="https://c/")) - assert len(FakeTool.instances) == 3 - # First instance (https://a/) was evicted → close() called. - assert FakeTool.instances[0].kwargs["url"] == "https://a/" - assert FakeTool.instances[0].close_count == 1 - # Other two remain in cache → not closed. - assert FakeTool.instances[1].close_count == 0 - assert FakeTool.instances[2].close_count == 0 - - @pytest.mark.asyncio - async def test_repeated_use_keeps_lru_alive(self) -> None: - handler = DefaultMCPToolHandler(cache_max_size=2) - with _patch_tool(): - await handler.invoke_tool(_invocation(server_url="https://a/")) - await handler.invoke_tool(_invocation(server_url="https://b/")) - # Touch a → b becomes LRU. - await handler.invoke_tool(_invocation(server_url="https://a/")) - # Insert c → b is evicted. - await handler.invoke_tool(_invocation(server_url="https://c/")) - # b was evicted. - b = FakeTool.instances[1] - assert b.kwargs["url"] == "https://b/" - assert b.close_count == 1 - # a survived. - a = FakeTool.instances[0] - assert a.kwargs["url"] == "https://a/" - assert a.close_count == 0 - - @pytest.mark.asyncio - async def test_concurrent_connect_shares_one_entry(self) -> None: - """Multiple concurrent invocations with the same key must share one tool.""" - handler = DefaultMCPToolHandler() - - # Slow down connect so concurrency window is observable. - original_connect = FakeTool.connect - - async def slow_connect(self: FakeTool) -> None: - self.connect_delay = 0.05 - await original_connect(self) - - with _patch_tool(), patch.object(FakeTool, "connect", slow_connect): - results = await asyncio.gather( - handler.invoke_tool(_invocation(headers={"X": "1"})), - handler.invoke_tool(_invocation(headers={"X": "1"})), - handler.invoke_tool(_invocation(headers={"X": "1"})), - handler.invoke_tool(_invocation(headers={"X": "1"})), - ) - assert all(not r.is_error for r in results) - # Only one tool was created and connected, despite 4 concurrent calls. - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].connect_count == 1 - - @pytest.mark.asyncio - async def test_different_connection_names_create_separate_entries(self) -> None: - """Same URL/headers but different ``connection_name`` must dispatch separately.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(connection_name="conn-A")) - await handler.invoke_tool(_invocation(connection_name="conn-B")) - assert len(FakeTool.instances) == 2 - - @pytest.mark.asyncio - async def test_different_server_labels_create_separate_entries(self) -> None: - """Same URL/headers but different ``server_label`` must dispatch separately.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(server_label="LabelA")) - await handler.invoke_tool(_invocation(server_label="LabelB")) - assert len(FakeTool.instances) == 2 - - @pytest.mark.asyncio - async def test_full_identity_match_hits_cache(self) -> None: - """All four identity components match → single cached entry.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(server_label="Lbl", connection_name="C", headers={"X": "1"})) - await handler.invoke_tool(_invocation(server_label="Lbl", connection_name="C", headers={"X": "1"})) - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].connect_count == 1 - - @pytest.mark.asyncio - async def test_header_name_case_collapses_to_one_cache_entry(self) -> None: - """Header name spelling differences (case-only) must share a cache entry.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"Authorization": "tk"})) - await handler.invoke_tool(_invocation(headers={"authorization": "tk"})) - await handler.invoke_tool(_invocation(headers={"AUTHORIZATION": "tk"})) - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].connect_count == 1 - - @pytest.mark.asyncio - async def test_header_value_case_does_not_collapse(self) -> None: - """Header *values* remain case-sensitive (different tokens → different sessions).""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"Authorization": "Bearer-A"})) - await handler.invoke_tool(_invocation(headers={"Authorization": "bearer-a"})) - assert len(FakeTool.instances) == 2 - - -# ---------- Aclose semantics ---------------------------------------------- - - -class TestAclose: - @pytest.mark.asyncio - async def test_aclose_closes_owned_clients(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"X": "1"})) - tool = FakeTool.instances[0] - owned = tool._httpx_client - assert owned is not None - await handler.aclose() - assert tool.close_count == 1 - assert owned.is_closed - - @pytest.mark.asyncio - async def test_aclose_does_not_close_caller_supplied_client(self) -> None: - caller_client = httpx.AsyncClient() - - async def provider(_inv: MCPToolInvocation) -> httpx.AsyncClient: - return caller_client - - handler = DefaultMCPToolHandler(client_provider=provider) - try: - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"X": "1"})) - await handler.aclose() - assert FakeTool.instances[0].close_count == 1 - # Caller client must still be usable. - assert not caller_client.is_closed - finally: - await caller_client.aclose() - - @pytest.mark.asyncio - async def test_async_context_manager(self) -> None: - with _patch_tool(): - async with DefaultMCPToolHandler() as handler: - await handler.invoke_tool(_invocation()) - tool = FakeTool.instances[0] - assert tool.close_count == 1 - - @pytest.mark.asyncio - async def test_aclose_is_idempotent(self) -> None: - """A second ``aclose`` is a no-op (no exception, no double-close).""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(headers={"X": "1"})) - await handler.aclose() - await handler.aclose() - assert FakeTool.instances[0].close_count == 1 - - @pytest.mark.asyncio - async def test_invoke_after_close_returns_error_result(self) -> None: - """Post-close ``invoke_tool`` surfaces a tool error rather than crashing.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.aclose() - result = await handler.invoke_tool(_invocation()) - assert result.is_error is True - assert "closed" in (result.error_message or "").lower() - - @pytest.mark.asyncio - async def test_aclose_drains_inflight_creation(self) -> None: - """An in-flight ``_create_entry`` must not leak when ``aclose`` races with it. - - Reproduces the race described in PR #5630 review-comment 3: - task A claims an inflight future and starts a slow connect; task B - runs ``aclose``; task A must self-clean (close its tool + httpx - client) and surface a closed-handler error rather than orphaning - the entry. - """ - handler = DefaultMCPToolHandler() - connect_started = asyncio.Event() - release_connect = asyncio.Event() - original_connect = FakeTool.connect - - async def gated_connect(self: FakeTool) -> None: - connect_started.set() - await release_connect.wait() - await original_connect(self) - - with _patch_tool(), patch.object(FakeTool, "connect", gated_connect): - invoke_task = asyncio.create_task(handler.invoke_tool(_invocation(headers={"X": "1"}))) - # Wait until task A is mid-connect. - await connect_started.wait() - # Race: kick off aclose. It must wait for the in-flight task. - close_task = asyncio.create_task(handler.aclose()) - # Yield once to ensure aclose has set _closed and is awaiting. - await asyncio.sleep(0) - # Allow the connect to complete; phase 3 sees _closed and self-cleans. - release_connect.set() - result = await invoke_task - await close_task - - # Entry was created and then closed by the in-flight task itself. - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].close_count == 1 - # The originating invocation surfaces a closed-handler error. - assert result.is_error is True - assert "closed" in (result.error_message or "").lower() - - -# ---------- Result normalisation ------------------------------------------ - - -class TestResultNormalisation: - @pytest.mark.asyncio - async def test_string_result_wrapped_in_text_content(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - inv = _invocation() - result = await handler.invoke_tool(inv) - # The fake's default already returns a list; replace handler for this test. - FakeTool.instances[0].call_handler = lambda **_a: "raw string body" - result = await handler.invoke_tool(inv) - assert result.is_error is False - assert len(result.outputs) == 1 - assert result.outputs[0].text == "raw string body" # type: ignore[reportAttributeAccessIssue] - - @pytest.mark.asyncio - async def test_list_result_passed_through(self) -> None: - handler = DefaultMCPToolHandler() - custom = [Content.from_text("a"), Content.from_text("b")] - with _patch_tool(): - inv = _invocation() - await handler.invoke_tool(inv) - FakeTool.instances[0].call_handler = lambda **_a: custom - result = await handler.invoke_tool(inv) - assert result.is_error is False - assert len(result.outputs) == 2 - - -# ---------- Error mapping -------------------------------------------------- - - -class TestErrorMapping: - @pytest.mark.asyncio - async def test_tool_execution_exception_returns_error_result(self) -> None: - handler = DefaultMCPToolHandler() - - def boom(**_a: Any) -> Any: - raise ToolExecutionException("server says no") - - with _patch_tool(): - inv = _invocation() - await handler.invoke_tool(inv) - FakeTool.instances[0].call_handler = boom - result = await handler.invoke_tool(inv) - assert result.is_error is True - assert result.error_message == "server says no" - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - assert text.startswith("Error:") - - @pytest.mark.asyncio - async def test_httpx_error_returns_error_result(self) -> None: - handler = DefaultMCPToolHandler() - - def boom(**_a: Any) -> Any: - raise httpx.ConnectError("dns failure") - - with _patch_tool(): - inv = _invocation() - await handler.invoke_tool(inv) - FakeTool.instances[0].call_handler = boom - result = await handler.invoke_tool(inv) - assert result.is_error is True - assert "dns failure" in (result.error_message or "") - - @pytest.mark.asyncio - async def test_unexpected_exception_propagates(self) -> None: - """RuntimeError (not in the narrow catch list) must propagate.""" - handler = DefaultMCPToolHandler() - - def boom(**_a: Any) -> Any: - raise RuntimeError("programmer error") - - with _patch_tool(): - inv = _invocation() - await handler.invoke_tool(inv) - FakeTool.instances[0].call_handler = boom - with pytest.raises(RuntimeError, match="programmer error"): - await handler.invoke_tool(inv) - - @pytest.mark.asyncio - async def test_connect_failure_returns_error_result(self) -> None: - handler = DefaultMCPToolHandler() - with ( - _patch_tool(), - patch.object( - FakeTool, - "connect", - lambda self: (_ for _ in ()).throw(httpx.ConnectError("server down")), - ), - ): - result = await handler.invoke_tool(_invocation()) - assert result.is_error is True - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - assert text.startswith("Error:") - # Failed connect must clear in-flight + cache entries. - assert handler._inflight == {} - assert len(handler._cache) == 0 - - @pytest.mark.asyncio - async def test_cancelled_error_propagates(self) -> None: - """asyncio.CancelledError is BaseException, must NOT be swallowed.""" - handler = DefaultMCPToolHandler() - - def boom(**_a: Any) -> Any: - raise asyncio.CancelledError - - with _patch_tool(): - inv = _invocation() - await handler.invoke_tool(inv) - FakeTool.instances[0].call_handler = boom - with pytest.raises(asyncio.CancelledError): - await handler.invoke_tool(inv) - - -# ---------- Cache key isolation ------------------------------------------- - - -class TestCacheKey: - def test_key_order_independent(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"A": "1", "B": "2"}) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"B": "2", "A": "1"}) - assert k1 == k2 - - def test_key_distinguishes_values(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"A": "1"}) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"A": "2"}) - assert k1 != k2 - - def test_empty_headers_use_fixed_hash(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, None, None) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {}) - assert k1 == k2 - - def test_key_distinguishes_connection_name(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, "conn-A", None) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, "conn-B", None) - assert k1 != k2 - - def test_key_distinguishes_server_label(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", "Lbl-A", None, None) - k2 = DefaultMCPToolHandler._cache_key("https://x/", "Lbl-B", None, None) - assert k1 != k2 - - def test_key_collapses_header_name_case(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"Authorization": "tk"}) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"authorization": "tk"}) - assert k1 == k2 - - def test_key_keeps_header_value_case(self) -> None: - k1 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"X": "Bearer-A"}) - k2 = DefaultMCPToolHandler._cache_key("https://x/", None, None, {"X": "bearer-a"}) - assert k1 != k2 - - -# ---------- tools/list reserved name -------------------------------------- - - -class TestListTools: - """Exercise the reserved :attr:`DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME` interception path.""" - - @pytest.mark.asyncio - async def test_list_tools_returns_json_catalog(self) -> None: - handler = DefaultMCPToolHandler() - with _patch_tool(): - # Prime the cache so the FakeTool session exists. - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session.list_tools_pages = [ # type: ignore[union-attr] # ty: ignore[invalid-assignment] - FakeListToolsResult( - tools=[ - FakeMcpTool( - name="search", - description="Search docs", - inputSchema={"type": "object", "properties": {"q": {"type": "string"}}}, - outputSchema={"type": "object"}, - ), - FakeMcpTool(name="echo", description=None, outputSchema=None), - ], - ), - ] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - assert result.is_error is False - assert len(result.outputs) == 1 - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - payload = json.loads(text) - assert payload == { - "tools": [ - { - "name": "search", - "description": "Search docs", - "inputSchema": {"type": "object", "properties": {"q": {"type": "string"}}}, - "outputSchema": {"type": "object"}, - }, - { - "name": "echo", - "description": None, - "inputSchema": {"type": "object", "properties": {}}, - "outputSchema": None, - }, - ], - } - - @pytest.mark.asyncio - async def test_list_tools_property_order_is_stable(self) -> None: - """JSON property order is stable: name, description, inputSchema, outputSchema.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session.list_tools_pages = [ # type: ignore[union-attr] # ty: ignore[invalid-assignment] - FakeListToolsResult(tools=[FakeMcpTool(name="t1", description="d")]), - ] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - name_idx = text.find('"name"') - desc_idx = text.find('"description"') - input_idx = text.find('"inputSchema"') - output_idx = text.find('"outputSchema"') - assert 0 <= name_idx < desc_idx < input_idx < output_idx - - @pytest.mark.asyncio - async def test_list_tools_indented_output(self) -> None: - """Output is JSON with a 2-space indent so the conversation log is human-readable.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session.list_tools_pages = [ # type: ignore[union-attr] # ty: ignore[invalid-assignment] - FakeListToolsResult(tools=[FakeMcpTool(name="t1")]), - ] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - # Indented output contains newlines and a 2-space indented key. - assert "\n " in text - - @pytest.mark.asyncio - async def test_list_tools_rejects_arguments(self) -> None: - """Reserved name does NOT accept tool arguments. Fails fast before connect.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - result = await handler.invoke_tool( - _invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME, arguments={"q": "test"}), - ) - assert result.is_error is True - assert "does not accept tool arguments" in (result.error_message or "") - # Args validation runs before connect, so no tool was instantiated. - assert FakeTool.instances == [] - - @pytest.mark.asyncio - async def test_list_tools_empty_args_dict_is_accepted(self) -> None: - """An empty arguments dict is equivalent to no arguments.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - result = await handler.invoke_tool( - _invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME, arguments={}), - ) - assert result.is_error is False - - @pytest.mark.asyncio - async def test_list_tools_paginates(self) -> None: - """Pagination loop calls list_tools repeatedly until nextCursor is empty.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session.list_tools_pages = [ # type: ignore[union-attr] # ty: ignore[invalid-assignment] - FakeListToolsResult(tools=[FakeMcpTool(name="a")], next_cursor="cursor1"), - FakeListToolsResult(tools=[FakeMcpTool(name="b")], next_cursor="cursor2"), - FakeListToolsResult(tools=[FakeMcpTool(name="c")], next_cursor=None), - ] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - text = result.outputs[0].text # type: ignore[reportAttributeAccessIssue] - assert text is not None - payload = json.loads(text) - assert [t["name"] for t in payload["tools"]] == ["a", "b", "c"] - session = FakeTool.instances[0].session - assert session is not None - assert len(session.list_tools_calls) == 3 - # First call has no cursor; second/third use the cursor from the prior page. - assert session.list_tools_calls[0] is None - assert getattr(session.list_tools_calls[1], "cursor", None) == "cursor1" - assert getattr(session.list_tools_calls[2], "cursor", None) == "cursor2" - - @pytest.mark.asyncio - async def test_list_tools_shares_cache_with_call_tool(self) -> None: - """tools/list reuses the same cached MCP session as a regular call_tool.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation(tool_name="search")) - await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - assert len(FakeTool.instances) == 1 - assert FakeTool.instances[0].connect_count == 1 - - @pytest.mark.asyncio - async def test_list_tools_propagates_session_errors_as_error_result(self) -> None: - """Errors raised by session.list_tools become MCPToolResult(is_error=True), not crashes.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session.list_tools_error = httpx.ReadTimeout("read timed out") # type: ignore[union-attr] # ty: ignore[invalid-assignment] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - assert result.is_error is True - assert "ReadTimeout" in (result.error_message or "") - - @pytest.mark.asyncio - async def test_list_tools_returns_error_when_session_is_none(self) -> None: - """If somehow the cached tool has no session, return a clear error rather than crashing.""" - handler = DefaultMCPToolHandler() - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].session = None - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - assert result.is_error is True - assert "not connected" in (result.error_message or "") - - @pytest.mark.asyncio - async def test_list_tools_does_not_call_call_tool(self) -> None: - """The reserved name is intercepted; the inner call_tool path is bypassed.""" - handler = DefaultMCPToolHandler() - call_tool_invoked = False - - def fail(**_a: Any) -> Any: - nonlocal call_tool_invoked - call_tool_invoked = True - raise AssertionError("call_tool should not run for tools/list") - - with _patch_tool(): - await handler.invoke_tool(_invocation()) - FakeTool.instances[0].call_handler = fail - FakeTool.instances[0].session.list_tools_pages = [ # type: ignore[union-attr] # ty: ignore[invalid-assignment] - FakeListToolsResult(tools=[]), - ] - result = await handler.invoke_tool(_invocation(tool_name=DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME)) - assert call_tool_invoked is False - assert result.is_error is False - - def test_class_attribute_value(self) -> None: - # Constant must equal the MCP protocol method name so a single - # string travels unchanged through host code, YAML, and the wire. - assert DefaultMCPToolHandler.LIST_TOOLS_TOOL_NAME == "tools/list" diff --git a/python/packages/declarative/tests/test_invoke_mcp_tool_executor.py b/python/packages/declarative/tests/test_invoke_mcp_tool_executor.py index b8c2e010c96..8e03b626f28 100644 --- a/python/packages/declarative/tests/test_invoke_mcp_tool_executor.py +++ b/python/packages/declarative/tests/test_invoke_mcp_tool_executor.py @@ -27,7 +27,8 @@ ) from agent_framework import Content, Message # noqa: E402 -from agent_framework.exceptions import ToolExecutionException # noqa: E402 +from agent_framework.exceptions import ToolException, ToolExecutionException # noqa: E402 +from mcp.shared.exceptions import MCPError # noqa: E402 from agent_framework_declarative._workflows import ( # noqa: E402 DECLARATIVE_STATE_KEY, @@ -604,26 +605,34 @@ async def test_handler_returns_error_result_assigns_error_string(self) -> None: decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY) assert decl["Local"]["Result"] == "Error: server down" - @pytest.mark.asyncio - async def test_tool_execution_exception_becomes_error_result(self) -> None: - handler = StubMcpHandler(raise_exc=ToolExecutionException("invalid arguments")) - factory = WorkflowFactory(mcp_tool_handler=handler) - workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"}))) - await workflow.run({}) - decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY) - assert decl["Local"]["Result"] == "Error: invalid arguments" + @pytest.mark.parametrize("require_approval", [False, True]) + @pytest.mark.parametrize( + "error", + [ + ToolException("connection failed"), + ToolExecutionException("tool failure"), + httpx.ConnectError("dns fail"), + MCPError(-32603, "protocol error"), + ], + ) + async def test_handler_exception_fails_workflow(self, error: Exception, require_approval: bool) -> None: + from agent_framework_declarative._workflows import ToolApprovalResponse - @pytest.mark.asyncio - async def test_httpx_error_becomes_error_result(self) -> None: - handler = StubMcpHandler(raise_exc=httpx.ConnectError("dns fail")) + handler = StubMcpHandler(raise_exc=error) factory = WorkflowFactory(mcp_tool_handler=handler) - workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"}))) - await workflow.run({}) - decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY) - result = decl["Local"]["Result"] - assert isinstance(result, str) - assert result.startswith("Error:") - assert "ConnectError" in result + workflow = factory.create_workflow_from_definition( + _yaml(_action(require_approval=require_approval, output={"result": "Local.Result"})) + ) + if require_approval: + paused = await workflow.run({}) + [request] = paused.get_request_info_events() + with pytest.raises(Exception) as excinfo: + await workflow.run(responses={request.request_id: ToolApprovalResponse(approved=True)}) + else: + with pytest.raises(Exception) as excinfo: + await workflow.run({}) + assert str(error) in str(excinfo.value) + assert handler.call_count == 1 @pytest.mark.asyncio async def test_unexpected_exception_propagates(self) -> None: diff --git a/python/samples/03-workflows/declarative/README.md b/python/samples/03-workflows/declarative/README.md index 9a24b741e77..92aa1091f79 100644 --- a/python/samples/03-workflows/declarative/README.md +++ b/python/samples/03-workflows/declarative/README.md @@ -67,6 +67,18 @@ actions: ### Tool Invocation - `InvokeFunctionTool` - Call a registered Python function +- `InvokeMcpTool` - Call an MCP tool through a host-supplied handler + +The `invoke_mcp_tool` and `invoke_foundry_toolbox_mcp` samples use +`DefaultMCPToolHandler`. Create this handler as a normal object; it has no +shutdown method or async context manager. Each invocation connects and closes +a fresh MCP protocol session, including calls after approval waits or checkpoint +resumes. Do not rely on implicit server-side session state between calls. + +The toolbox sample supplies a caller-owned HTTP client through `client_provider` +to reuse its authenticated HTTP connection pool without sharing MCP sessions. +Keep that client's context open until every workflow call has finished; cancel +and await outstanding calls before closing it. ### Human-in-Loop - `Question` - Request user input diff --git a/python/samples/03-workflows/declarative/invoke_foundry_toolbox_mcp/main.py b/python/samples/03-workflows/declarative/invoke_foundry_toolbox_mcp/main.py index 55765b41707..6bc0a2ccfd0 100644 --- a/python/samples/03-workflows/declarative/invoke_foundry_toolbox_mcp/main.py +++ b/python/samples/03-workflows/declarative/invoke_foundry_toolbox_mcp/main.py @@ -14,6 +14,8 @@ python samples/03-workflows/declarative/invoke_foundry_toolbox_mcp/main.py """ +from __future__ import annotations + import asyncio import os from collections.abc import Generator @@ -99,13 +101,10 @@ async def _client_provider(invocation: MCPToolInvocation) -> httpx.AsyncClient | ) return http_client - async with ( - http_client, - DefaultMCPToolHandler(client_provider=_client_provider) as mcp_handler, - ): + async with http_client: factory = WorkflowFactory( agents={AGENT_NAME: summary_agent}, - mcp_tool_handler=mcp_handler, + mcp_tool_handler=DefaultMCPToolHandler(client_provider=_client_provider), configuration={ "FOUNDRY_TOOLBOX_MCP_SERVER_URL": toolbox_endpoint, "FOUNDRY_TOOLBOX_DOCS_SERVER_LABEL": DOCS_SERVER_LABEL, diff --git a/python/samples/03-workflows/declarative/invoke_mcp_tool/main.py b/python/samples/03-workflows/declarative/invoke_mcp_tool/main.py index 358ee919047..51e06a6a7dc 100644 --- a/python/samples/03-workflows/declarative/invoke_mcp_tool/main.py +++ b/python/samples/03-workflows/declarative/invoke_mcp_tool/main.py @@ -39,6 +39,8 @@ MCP_REQUIRE_APPROVAL=1 python -m samples.03-workflows.declarative.invoke_mcp_tool.main """ +from __future__ import annotations + import asyncio import os from pathlib import Path @@ -127,76 +129,75 @@ async def main() -> None: # requests through a pre-configured ``httpx.AsyncClient`` carrying the # appropriate credentials, or wrap the handler with one that injects # headers per call. - async with DefaultMCPToolHandler() as mcp_handler: - factory = WorkflowFactory( - agents=agents, - mcp_tool_handler=mcp_handler, - ) - - workflow_path = Path(__file__).parent / "workflow.yaml" - workflow = factory.create_workflow_from_yaml_path(workflow_path) - - print("=" * 60) - print("Invoke MCP Tool Workflow Demo") - if require_approval: - print("(MCP_REQUIRE_APPROVAL is set — you will be prompted before the tool runs)") + factory = WorkflowFactory( + agents=agents, + mcp_tool_handler=DefaultMCPToolHandler(), + ) + + workflow_path = Path(__file__).parent / "workflow.yaml" + workflow = factory.create_workflow_from_yaml_path(workflow_path) + + print("=" * 60) + print("Invoke MCP Tool Workflow Demo") + if require_approval: + print("(MCP_REQUIRE_APPROVAL is set — you will be prompted before the tool runs)") + else: + print("(set MCP_REQUIRE_APPROVAL=1 to enable the human-approval flow)") + print("=" * 60) + print() + print("Ask one question that can be answered from the Microsoft Learn docs or provide a keyword to search.") + print() + + user_input = input("You: ").strip() # noqa: ASYNC250 + if not user_input: + user_input = "What is the Agent Framework declarative workflow runtime?" + + # Drive the workflow via dict-shaped inputs so the YAML can read + # both the user's question (``Workflow.Inputs.text``) and the + # approval toggle (``Workflow.Inputs.requireApproval``) without + # any Python-side mutation of the workflow definition. + workflow_inputs: dict[str, object] = { + "text": user_input, + "requireApproval": require_approval, + } + + # The request_info loop below handles the MCP approval flow when + # the YAML requests it. When ``requireApproval`` is false the + # workflow never emits an ``MCPToolApprovalRequest`` event, so + # the loop runs exactly once and exits cleanly — both modes share + # the same code path. + pending: tuple[str, MCPToolApprovalRequest] | None = None + produced_output = False + printed_agent_prefix = False + + while True: + if pending is None: + stream = workflow.run(workflow_inputs, stream=True) else: - print("(set MCP_REQUIRE_APPROVAL=1 to enable the human-approval flow)") - print("=" * 60) - print() - print("Ask one question that can be answered from the Microsoft Learn docs or provide a keyword to search.") - print() - - user_input = input("You: ").strip() # noqa: ASYNC250 - if not user_input: - user_input = "What is the Agent Framework declarative workflow runtime?" - - # Drive the workflow via dict-shaped inputs so the YAML can read - # both the user's question (``Workflow.Inputs.text``) and the - # approval toggle (``Workflow.Inputs.requireApproval``) without - # any Python-side mutation of the workflow definition. - workflow_inputs: dict[str, object] = { - "text": user_input, - "requireApproval": require_approval, - } - - # The request_info loop below handles the MCP approval flow when - # the YAML requests it. When ``requireApproval`` is false the - # workflow never emits an ``MCPToolApprovalRequest`` event, so - # the loop runs exactly once and exits cleanly — both modes share - # the same code path. - pending: tuple[str, MCPToolApprovalRequest] | None = None - produced_output = False - printed_agent_prefix = False - - while True: - if pending is None: - stream = workflow.run(workflow_inputs, stream=True) + pending_id, pending_request = pending + response = _prompt_for_approval(pending_request) + stream = workflow.run(stream=True, responses={pending_id: response}) + pending = None + + async for event in stream: + if event.type == "output" and isinstance(event.data, str): + if not printed_agent_prefix: + print("\nAgent: ", end="", flush=True) + printed_agent_prefix = True + print(event.data, end="", flush=True) + produced_output = True + elif event.type == "request_info" and isinstance(event.data, MCPToolApprovalRequest): + pending = (event.request_id, event.data) + + if pending is None: + if not produced_output: + # Workflow finished without producing any agent output + # (e.g. the user rejected the MCP tool call and the + # downstream agent had nothing to summarise). + print("\n(no response produced)") else: - pending_id, pending_request = pending - response = _prompt_for_approval(pending_request) - stream = workflow.run(stream=True, responses={pending_id: response}) - pending = None - - async for event in stream: - if event.type == "output" and isinstance(event.data, str): - if not printed_agent_prefix: - print("\nAgent: ", end="", flush=True) - printed_agent_prefix = True - print(event.data, end="", flush=True) - produced_output = True - elif event.type == "request_info" and isinstance(event.data, MCPToolApprovalRequest): - pending = (event.request_id, event.data) - - if pending is None: - if not produced_output: - # Workflow finished without producing any agent output - # (e.g. the user rejected the MCP tool call and the - # downstream agent had nothing to summarise). - print("\n(no response produced)") - else: - print() - break + print() + break if __name__ == "__main__": diff --git a/skills/agent-framework-builder-python/references/declarative.md b/skills/agent-framework-builder-python/references/declarative.md index ce46be674cf..f5a0a2cc59e 100644 --- a/skills/agent-framework-builder-python/references/declarative.md +++ b/skills/agent-framework-builder-python/references/declarative.md @@ -24,6 +24,14 @@ maturity. Check current package status and public docstrings before making compa - Allowlist external destinations and register only resources the definition may invoke. - Review the documented HTTP body limits and handler ownership rules; use a custom handler when host policy requires different authentication, retries, SSRF controls, or transport behavior. +- `DefaultMCPToolHandler` opens and closes a fresh MCP session per invocation, including `tools/list`. + Instantiate it normally, not with `async with`; it has no `cache_max_size` or `aclose()`. + Its optional `client_provider` may return a borrowed, reusable HTTP client, but protocol state is never shared. + Cancel and await active calls before closing borrowed clients. Implicit server state does not survive later calls, + approval waits, or checkpoint resumes; use explicit remote state handles or a custom scoped handler if needed. + Setup, connection, discovery, and transport exceptions fail the workflow action, including after approval. + Remote tool error responses remain error results; custom handlers must raise infrastructure failures rather than + returning them as tool output. ## Validation