Working MCP client tools with openai azure hosted models

[alexbozhkov]

Description

Hello, this is a working solution, for integrating MCP client with openai azure hosted models that were incompatible.

I tried a basic implementation according to the pydantic-ai MCP client documentation:

server = MCPServerStreamableHTTP(url=settings.MCP_URL)

def _build_agent() -> Agent:
    model = OpenAIModel(
        model_name=settings.MINI_MODEL_NAME,
        provider=AzureProvider(
            azure_endpoint=settings.AZURE_ENDPOINT,
            api_version=settings.API_VERSION,
            api_key=settings.API_KEY,
        ),
    )

    # Optional: Demonstrate history_processors usage with a simple limiter
    def keep_recent(messages: List[ModelMessage]) -> List[ModelMessage]:
        max_messages = settings.CHAT_HISTORY_MAX_MESSAGES
        if max_messages is None:
            return messages
        return messages[-max_messages:]

    return Agent(
        model=model,
        toolsets=[server],
        history_processors=[keep_recent],
    )

But I was getting the following errors:

• "Invalid parameter: messages with role 'tool' must be a response to a preceeding message with 'tool_calls'.", 'type': 'invalid_request_error'
• TypeError: Cannot instantiate typing.Union for the ChatCompletionMessageToolCallParam.

This is how I've integrated it in my fastapi app and it successfully ran the mcp tools:

def install_openai_sdk_compat() -> None:
    """Install a runtime adapter for OpenAI SDK ≥ 1.90 tool_call params.

    Newer OpenAI SDKs define `ChatCompletionMessageToolCallParam` as a `TypeAlias` (Union),
    which is not callable. pydantic-ai 0.6.2 tries to instantiate it, causing
    `TypeError: Cannot instantiate typing.Union`.

    This shim makes that symbol callable by returning the required dict shape
    and wraps `OpenAIModel._map_tool_call` to fall back to a dict if anything fails.
    """
    try:
        import pydantic_ai.models.openai as _pai_openai  # type: ignore
        from openai.types import chat as _oai_chat

        ToolCallParam = getattr(_oai_chat, "ChatCompletionMessageToolCallParam", None)
        if ToolCallParam is not None and not callable(ToolCallParam):

            def _compat_tool_param(*, id: str, type: str, function: dict[str, Any]):
                return {"id": id, "type": type, "function": function}

            _pai_openai.chat.ChatCompletionMessageToolCallParam = _compat_tool_param  # type: ignore[attr-defined]

        _orig = _pai_openai.OpenAIModel._map_tool_call

        def _map_tool_call_compat(self, t):  # type: ignore[no-redef]
            try:
                return _orig(self, t)
            except Exception:
                return {
                    "id": getattr(t, "tool_call_id", ""),
                    "type": "function",
                    "function": {
                        "name": t.tool_name,
                        "arguments": t.args_as_json_str(),
                    },
                }

        _pai_openai.OpenAIModel._map_tool_call = _map_tool_call_compat  # type: ignore[assignment]
    except Exception:
        try:
            import pydantic_ai.models.openai as _pai_openai  # type: ignore

            def _compat_tool_param(*, id: str, type: str, function: dict[str, Any]):
                return {"id": id, "type": type, "function": function}

            _pai_openai.chat.ChatCompletionMessageToolCallParam = _compat_tool_param  # type: ignore[attr-defined]

            def _map_tool_call_compat(self, t):  # type: ignore[no-redef]
                return {
                    "id": getattr(t, "tool_call_id", ""),
                    "type": "function",
                    "function": {
                        "name": t.tool_name,
                        "arguments": t.args_as_json_str(),
                    },
                }

            _pai_openai.OpenAIModel._map_tool_call = _map_tool_call_compat  # type: ignore[assignment]
        except Exception:
            # Give up silently; better to proceed than crash import
            pass


install_openai_sdk_compat()

server = MCPServerStreamableHTTP(url=settings.MCP_URL)


def _build_agent() -> Agent:
    model = OpenAIModel(
        model_name=settings.MINI_MODEL_NAME,
        provider=AzureProvider(
            azure_endpoint=settings.AZURE_ENDPOINT,
            api_version=settings.API_VERSION,
            api_key=settings.API_KEY,
        ),
    )
    def keep_recent(messages: List[ModelMessage]) -> List[ModelMessage]:
        max_messages = settings.CHAT_HISTORY_MAX_MESSAGES
        if max_messages is None:
            return messages
        return messages[-max_messages:]

    return Agent(
        model=model,
        toolsets=[server],
        history_processors=[keep_recent],
    )

These are the packages I use:

fastapi[standard]==0.115.11
pre-commit==4.1.0
pytest-asyncio==0.25.3
pytest-cov==6.0.0
pydantic-settings==2.8.1
pydantic-ai==0.6.2
uvicorn==0.29.0

References

No response

[DouweM]

@alexbozhkov The TypeError was already tracked in #2476 and appears to have been fixed in the latest version of the openai package.