Skip to content

Python: [Bug]: MCP tools return double output if CallToolResult contains both content and structuredContent #7866

Description

Description

  • What happened? Agent called MCP server tool but response visible to the Agent was duplicated: one was content, the other was structuredContent. Occurred when using my own MCP server, but also tested on MS Learn MCP and Deep Wiki MCP servers.
  • What I expected: no duplication, but structured content.
  • Directly call MS Learn MCP / Deep Wiki MCP via call_tool and see results.

Code Sample

import asyncio

from agent_framework import MCPStreamableHTTPTool

# DeepWiki's server (built on fastmcp) tags each tool's `_meta` with a `_fastmcp`
# key, whose leading underscore violates the MCP spec's _meta key-name format.
# agent_framework enforces that format strictly and aborts tool loading over it,
# so relax it here to reach the server for this repro. Separate bug, unrelated
# to the duplicate-content issue below.
import agent_framework._mcp as _af_mcp

_af_mcp._validate_mcp_meta_key = lambda key: None

DEEPWIKI_MCP_URL = "https://mcp.deepwiki.com/mcp"
TOOL_NAME = "ask_question"
TOOL_ARGS = {"repoName": "microsoft/agent-framework", "question": "What is this repository about?"}


async def main() -> None:
    async with MCPStreamableHTTPTool(name="DeepWiki", url=DEEPWIKI_MCP_URL, load_prompts=False) as deepwiki:
        result = await deepwiki.call_tool(TOOL_NAME, **TOOL_ARGS)
        print(f"call_tool() returned {len(result)} content item(s) (expected 1)")
        for i, item in enumerate(result):
            print(f"  item[{i}]: {getattr(item, 'text', item)!r:.200}")

        # Bypass agent_framework's parsing to inspect the raw CallToolResult:
        # confirms the server sends exactly one `content` block plus
        # `structuredContent` -- the duplication is manufactured client-side.
        raw = await deepwiki.session.call_tool(TOOL_NAME, arguments=TOOL_ARGS)
        print(f"raw.content: {len(raw.content)} block(s); raw.structuredContent is None: {raw.structuredContent is None}")


if __name__ == "__main__":
    asyncio.run(main())

Error Messages / Stack Traces

uv run python repro_mcp_duplicate_content.py
call_tool() returned 2 content item(s) (expected 1)
  item[0]: 'This repository, `microsoft/agent-framework`, is a multi-language framework designed for building, orchestrating, and deploying AI agents . It provides parallel implementations in both .NET and Pytho
  item[1]: '{"result": "This repository, `microsoft/agent-framework`, is a multi-language framework designed for building, orchestrating, and deploying AI agents . It provides parallel implementations in both .N
raw.content: 1 block(s); raw.structuredContent is None: False

Package Versions

1.13.0

Python Version

3.12

Additional Context

Probable root cause: agent_framework/_mcp.py::_parse_tool_result_from_mcp unconditionally
appends a second Content item built from structuredContent, even when content
already carries an equivalent serialization.
So even if for-loop succeeds appending content, next if block will append structuredContent anyway.

    def _parse_tool_result_from_mcp(
        self,
        mcp_type: types.CallToolResult,
    ) -> list[Content]:
        """Parse an MCP CallToolResult into a list of Content items.

        If the server attached a ``_meta`` payload to the tool result (e.g. for
        Information Flow Control labels under the ``ifc`` key), a copy of that
        payload is stamped onto each produced :class:`Content` instance under
        ``additional_properties["_meta"]``.  Downstream layers (such as
        :class:`agent_framework.security.SecureMCPToolProxy`) consume this key
        to derive per-item security labels.
        The sentinel is intentionally generic so any MCP server's ``_meta``
        keys (current or future) can be interpreted by higher-level code.
        """
        from mcp import types

        raw_meta = mcp_type.meta
        meta: dict[str, Any] | None = dict(raw_meta) if isinstance(raw_meta, Mapping) else None
        # Stamp the server ``_meta`` payload directly via additional_properties on
        # each newly constructed Content; empty when the server provided no meta.
        additional_kwargs: dict[str, Any] = {"additional_properties": {"_meta": meta}} if meta else {}

        result: list[Content] = []
        for item in mcp_type.content:
            match item:
                case types.TextContent():
                    result.append(Content.from_text(item.text, **additional_kwargs))
                case types.ImageContent() | types.AudioContent():
                    decoded = base64.b64decode(item.data)
                    result.append(
                        Content.from_data(
                            data=decoded,
                            media_type=item.mimeType,
                            **additional_kwargs,
                        )
                    )
                case types.ResourceLink():
                    result.append(
                        Content.from_uri(
                            uri=str(item.uri),
                            media_type=item.mimeType,
                            **additional_kwargs,
                        )
                    )
                case types.EmbeddedResource():
                    match item.resource:
                        case types.TextResourceContents():
                            result.append(Content.from_text(item.resource.text, **additional_kwargs))
                        case types.BlobResourceContents():
                            blob = item.resource.blob
                            mime = item.resource.mimeType or "application/octet-stream"
                            if not blob.startswith("data:"):
                                blob = f"data:{mime};base64,{blob}"
                            result.append(
                                Content.from_uri(
                                    uri=blob,
                                    media_type=mime,
                                    **additional_kwargs,
                                )
                            )
                case _:
                    result.append(Content.from_text(str(item), **additional_kwargs))

        if mcp_type.structuredContent is not None:
            result.append(Content.from_text(json.dumps(mcp_type.structuredContent, default=str)))

        if not result:
            result.append(Content.from_text("null", **additional_kwargs))
        return result

Metadata

Metadata

Labels

.NETUsage: [Issues, PRs], Target: .NetagentsUsage: [Issues, PRs], Target: Single agentmcpUsage: [Issues, PRs], Target: MCPpythonUsage: [Issues, PRs], Target: PythonreproducedUsage: [Issues], Target: all issues that can be reproduced by the triage workflow

Type

Projects

Status
No status

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions