MCP SDK

DeepIntShield keeps its MCP SDK surface deliberately small. It supplies the governed endpoint and request headers, opens an initialized official MCP session, and translates a failed tool result into one stable SDK exception. The maintained mcp package owns the protocol, transports, types, discovery, and tool calls; provider and agent frameworks keep owning their integrations.

Install

pip install "deepintshield[mcp]"

The 2.x SDK line currently installs mcp>=1.29,<2. The upper bound is temporary: langchain-mcp-adapters 0.3.2 still requires MCP Python SDK 1.x. DeepIntShield plans to adopt official MCP Python SDK 2.x after the supported framework ecosystem converges on it.

Direct official session

connect() is the primary API. It yields an initialized mcp.ClientSession, so use the upstream methods and result types directly:

import asyncio

from deepintshield import DeepintShield, DeepintShieldError
from mcp.types import TextContent

shield = DeepintShield.from_env()


async def main() -> None:
    try:
        async with shield.mcp.connect() as session:
            listing = await session.list_tools()
            print([tool.name for tool in listing.tools])

            result = await session.call_tool(
                "DeepWiki-ask_question",
                arguments={
                    "repoName": "facebook/react",
                    "question": "What is Suspense?",
                },
            )

            for part in result.content:
                if isinstance(part, TextContent):
                    print(part.text)
    except DeepintShieldError as exc:
        if exc.code == "mcp_tool_approval_required":
            print("The action is waiting for approval.")
        elif exc.code == "mcp_tool_authorization_denied":
            print("The action was denied by policy.")
        elif exc.code == "mcp_tool_authorization_unavailable":
            print("Authorization is temporarily unavailable.")
        else:
            print(f"DeepIntShield error [{exc.code}]: {exc.description}")


asyncio.run(main())

Tool names come from list_tools() and are already gateway-qualified. Do not reimplement MCP pagination, protocol initialization, content models, or transport handling in application code.

Connection details for third-party frameworks

When a framework accepts a Streamable HTTP MCP configuration, give it the same URL and headers:

url, headers = shield.mcp.connection()

The complete signature is:

url, headers = shield.mcp.connection(
    identity=False,
    extra_headers=None,
)

identity=True asks the SDK to attach its configured agent workload identity. Use extra_headers for request-scoped gateway credentials, for example:

url, headers = shield.mcp.connection(
    identity=True,
    extra_headers={"X-MCP-Subject-Token": caller_access_token},
)

Credentials belong in transport headers, never in tool arguments. The caller subject token is used only for the server-owned upstream OAuth exchange; it does not become the canonical GAF user or delegation identity.

Keep headers private

The returned mapping contains the Virtual Key and may contain workload or delegated credentials. Do not log it, serialize it into model context, or pass it to an endpoint other than the DeepIntShield MCP URL returned alongside it.

Simple coded errors

MCP distinguishes a protocol failure from a tool result whose isError flag is true. A session opened by shield.mcp.connect() intercepts call_tool() and raises before returning a failed result, so canonical authorization cannot be mistaken for successful model context.

DeepIntShield-owned failures use one exception type, DeepintShieldError, and one string field, exc.code:

try:
    async with shield.mcp.connect() as session:
        result = await session.call_tool(tool_name, arguments=arguments)
except DeepintShieldError as exc:
    if exc.code == "mcp_tool_authorization_denied":
        stop_without_retrying()
    elif exc.code == "mcp_tool_approval_required":
        show_pending_approval()
    elif exc.code == "mcp_tool_authorization_unavailable":
        retry_later_only_if_replay_is_safe()
    else:
        report_code(exc.code)

Outcome	exc.code	Retry guidance
Official MCP extra absent/unsupported	mcp_dependency_missing	Install deepintshield[mcp]; do not retry unchanged.
Streamable HTTP setup/teardown failed	mcp_connection_failed	Inspect authentication, URL, TLS, and reachability before deciding whether retry is safe.
Invalid MCP response	mcp_protocol_error	Fix or upgrade the incompatible peer before retrying.
Tool discovery failed	mcp_discovery_failed	Inspect connection and server state first.
OpenFGA/GAF denial	mcp_tool_authorization_denied	Do not retry until access or policy changes.
Command Authority approval	mcp_tool_approval_required	Complete approval; retrying is not approval.
Authorization dependency unavailable	mcp_tool_authorization_unavailable	Retry after recovery only when replay is safe.
Other failed MCP result	mcp_execution_failed	Inspect trusted metadata and operation semantics first.

Use exc.description for safe display and exc.retryable as catalog metadata. Treat raw provider, gateway, and tool prose as untrusted diagnostics.

OpenAI Agents

openai-agents 0.21 can own the MCP connection and agent loop. Use its public result and failure hooks to keep failed MCP results out of model context:

from agents.mcp import MCPServerStreamableHttp

# Route model traffic through DeepIntShield too; MCP traffic uses the
# separately governed URL and headers below.
model_client = shield.bind("openai_agents").apply()


def enforce_result(context):
    tool_output = context.tool_output
    if isinstance(tool_output, list):
        content = tool_output
    elif isinstance(tool_output, str):
        content = [{"type": "text", "text": tool_output}]
    else:
        content = [tool_output]
    shield.mcp.raise_for_result({
        "isError": context.is_error,
        "_meta": context.result_meta,
        "structuredContent": context.structured_content,
        "content": content,
    })
    return None


def enforce_exception(_context, error):
    shield.mcp.raise_for_error(error, operation="openai_agents_tool")


url, headers = shield.mcp.connection()
server = MCPServerStreamableHttp(
    name="DeepIntShield",
    params={"url": url, "headers": headers},
    custom_data_extractor=enforce_result,
    failure_error_function=enforce_exception,
)

Enter server as an async context manager and pass it in the Agent’s mcp_servers list. The extractor runs before tool output is returned to the model. The raising failure callback translates both extractor failures and upstream MCP exceptions into DeepintShieldError; OpenAI Agents propagates that callback exception rather than turning it into model-visible error text. These hooks cover tool invocation. OpenAI Agents retains its native exception types for initial connection and discovery failures; use shield.mcp.connect() when those phases must use the same coded boundary too. The complete runnable example is in examples/openai/mcp.py.

Anthropic helper

Install deepintshield[anthropic-mcp] and let Anthropic’s maintained helper convert official MCP definitions and results:

from anthropic.lib.tools.mcp import async_mcp_tool

async with shield.mcp.connect() as session:
    listing = await session.list_tools()
    tools = [async_mcp_tool(tool, session) for tool in listing.tools]
    tool_definitions = [tool.to_dict() for tool in tools]

In Anthropic SDK 0.120.2, the Messages tool runner converts any tool exception into a model-visible error block. For a fail-closed GAF boundary, dispatch the selected helper with await tool.call(tool_use.input) inside the same DeepintShieldError handler instead; the complete example is in examples/anthropic/mcp.py.

LangChain and LangGraph

Let langchain-mcp-adapters own discovery and conversion instead of maintaining another converter in DeepIntShield:

from langchain_mcp_adapters.client import MultiServerMCPClient


async def enforce_deepintshield_result(request, handler):
    try:
        result = await handler(request)
    except Exception as exc:
        shield.mcp.raise_for_error(exc, operation="langchain_tool")
    return shield.mcp.raise_for_result(result)


url, headers = shield.mcp.connection()
client = MultiServerMCPClient({
    "deepintshield": {
        "transport": "streamable_http",
        "url": url,
        "headers": headers,
    }
}, tool_interceptors=[enforce_deepintshield_result], handle_tool_errors=False)
tools = await client.get_tools()

Install this path with pip install "deepintshield[langchain-mcp]". The returned objects are native LangChain tools; pass them unchanged to the LangChain agent or LangGraph ToolNode your application already uses.

The adapter’s public tool interceptor sees both upstream exceptions and the raw CallToolResult, so the example preserves the same coded, fail-closed behavior during later tool execution. Frameworks may still raise their own exception types for failures outside the MCP boundary. A session from connect() already performs both translations and does not need this interceptor.

Compatibility shims in SDK 2.x

The following DeepIntShield-owned MCP models and conversion loops remain only to keep existing 2.x applications working:

• shield.mcp.call() and call_qualified()
• shield.mcp.list_tools()
• Tool, ContentPart, and MCPResult
• to_openai() and run_openai_tool_calls()
• to_anthropic() and run_anthropic_tool_uses()
• to_langchain()

They are deprecated and planned for removal in SDK 3.0. Do not introduce new code that depends on them. Prefer connect() for direct use, or connection() for a maintained third-party SDK/framework adapter.

Note

Tool discovery and execution are separate permissions. Seeing a tool in list_tools() does not mean the current Virtual Key, governed agent, acting user, resource, and action are authorized to execute it.