Python: [Feature]: Python MCP example for schema-validated remote field extraction

Description

Would a small Python MCP sample that validates a remote extraction request against the discovered input schema, then requires explicit opt-in before one paid call, be useful in python/samples/02-agents/mcp/? I would like to implement and maintain this sample if the scope fits.

The current README already covers anonymous search/fetch and keyed Serply search (merged in #8319). This proposal teaches a different step: inspect the live tool contract, request named fields from a public page, validate the arguments before invoking the tool, and make the cost-bearing step explicit. It uses the existing MCPStreamableHTTPTool API and needs no new framework API or model provider account.

The proposed service is Baizhi Agent Toolkit, a commercial hosted MCP service authenticated with a user-supplied static Bearer token. Its backend is closed source; the public integration repository contains client/documentation material, not the server implementation. Tool requests may consume paid credits. The example would state those facts and send only a fixed public documentation URL. BAIZHI_API_KEY would be supplied through the user's environment secret manager.

Proposed scope:

• One self-contained sample and a short entry in the existing MCP README.
• Use static_headers for the fixed token and expose only web_extract with allowed_tools; the latter narrows client functions, not server-side permissions or billing.
• Inspect the discovered schema and validate the proposed arguments before tools/call. The provider's fields value is a map such as {"title": "string", "summary": "string"}, not an MCP outputSchema or a guaranteed output type.
• Default to discovery only. An explicit --run sends one extraction request with download=False; no application retry loop. Discovery still contacts the service, so the sample would not claim that discovery is free.

The request shape below comes from an earlier authenticated tools/list capture at 2026-09-20 07:59:08 UTC; that capture made no tools/call requests. The endpoint and Bearer setup are documented in the official setup guide. This proposal has not made a new live discovery or a production extraction call and does not claim service output/quality or billing has been validated.

Local verification used the published agent-framework-core==1.19.0, mcp==1.30.0, and jsonschema==4.26.0 distributions on Python 3.12, not a checkout build. Nine tests passed with the real Agent Framework and MCP client over httpx.MockTransport, synthetic credentials, and no outbound network. They cover zero calls by default, exactly one explicit call, missing/changed tool schema, tool errors, authentication failures, empty credentials, and rejected cross-origin redirects. The candidate sample also passes targeted Ruff and Pyright checks; the full repository checks and production service acceptance have not been run.

If the existing MCP samples already cover enough of this workflow, I can keep the service-specific example in the public client repository instead. Is this narrow extraction/schema-validation example in scope, and would maintainers prefer a generic local-server example?

Disclosure: I am submitting this as part of the Baizhi Agent Toolkit integration effort. AI assistance was used to prepare the candidate code, tests, and this proposal. Maintainer agreement is being requested before opening a code PR.

Code Sample

The following is the locally tested candidate, pending scope review:

# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "agent-framework-core==1.19.0",
#     "mcp==1.30.0",
#     "jsonschema==4.26.0",
# ]
# ///
# Copyright (c) Microsoft. All rights reserved.

import argparse
import asyncio
import json
import os

from agent_framework import MCPStreamableHTTPTool
from jsonschema import validate

"""
Inspect an authenticated remote MCP extraction tool before opting into one paid call.

Required environment variable: BAIZHI_API_KEY (a Baizhi Agent Toolkit API key).
Run without arguments to discover the schema; use --run to request extraction.
The hosted service is commercial and its backend is not open source.
No model provider account is needed because this example calls the MCP tool directly.
"""


async def main(run: bool = False) -> None:
    """Inspect an extraction tool, optionally making one explicit paid request."""
    # 1. Read a user-supplied secret without placing it in tool arguments.
    api_key = os.environ["BAIZHI_API_KEY"]
    if not api_key.strip():
        raise ValueError("Set BAIZHI_API_KEY through your environment secret manager.")

    # 2. Expose only the extraction function and attach a fixed authentication header.
    async with MCPStreamableHTTPTool(
        name="baizhi-extraction",
        url="https://agent-toolkit.app.baizhi.cloud/mcp",
        static_headers={"Authorization": f"Bearer {api_key}"},
        allowed_tools=["web_extract"],
        load_prompts=False,
        request_timeout=30,
        parse_tool_results=lambda result: "\n".join(c.text for c in result.content if c.type == "text"),
    ) as mcp:
        extract = next((tool for tool in mcp.functions if tool.name == "web_extract"), None)
        if extract is None:
            raise RuntimeError("web_extract is unavailable; inspect the connected server before continuing.")

        # 3. Inspect the live input contract instead of guessing the provider's fields.
        schema = extract.parameters()
        print(json.dumps(schema, ensure_ascii=False, indent=2))
        arguments = {
            "url": "https://learn.microsoft.com/en-us/agent-framework/overview/",
            "fields": {"title": "string", "summary": "string"},
            "instruction": "Extract the page title and a one-sentence summary from this public documentation page.",
            "download": False,
        }
        validate(instance=arguments, schema=schema)
        if not run:
            print("Discovery only: no tools/call sent. Use --run only after reviewing the service's current charges.")
            return

        # 4. Field type names belong to this provider's input, not an MCP outputSchema.
        result = await mcp.call_tool("web_extract", **arguments)
        print(result)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Inspect a remote field-extraction tool before opting into a paid call."
    )
    parser.add_argument("--run", action="store_true", help="Send the extraction request; paid credits may be consumed.")
    asyncio.run(main(parser.parse_args().run))

# Expected output:
# The current extraction input schema, followed by the discovery-only message.
# With --run, the server's extraction result is printed instead; its format may vary.

Language/SDK

Python