Implementation:Openai Openai agents python HostedMCP Connector Pattern

Knowledge Sources	Openai_Openai_agents_python
Domains	MCP Integration, OAuth Authentication, Tool Connectors
Last Updated	2026-02-11 00:00 GMT

Overview

Demonstrates using HostedMCPTool with an OpenAI-hosted connector (Google Calendar) that requires OAuth authorization, including both synchronous and streaming execution modes with event handling.

Description

The HostedMCP Connector pattern enables agents to use pre-built connectors hosted on the OpenAI platform that integrate with external services. Unlike self-hosted MCP servers, connectors are managed by OpenAI and referenced by a connector_id. This example uses the Google Calendar connector (connector_googlecalendar) to allow an agent to query a user's calendar.

The connector requires OAuth authorization, provided via an access token obtained from the Google OAuth Playground. The token is passed in the authorization field of the tool_config dictionary. The require_approval field is set to "never", meaning the agent can invoke the MCP tools without user confirmation. The server_label field provides a human-readable identifier for the connector.

The example supports both non-streaming and streaming modes. In streaming mode, it iterates over run_result.stream_events() and filters for specific event types: response.output_item events for structured output items, response.mcp events for MCP-specific interactions, and response.output_text.delta events for incremental text output. A verbose mode is also available that prints all new_items from the run result for debugging purposes.

Usage

Use this pattern when integrating agents with external services via OpenAI-hosted connectors that handle protocol translation and authentication. This is ideal for scenarios like calendar management, email access, or other third-party API integrations where the connector abstraction simplifies the integration compared to self-hosted MCP servers.

Code Reference

Source Location

• Repository: Openai_Openai_agents_python
• File: examples/hosted_mcp/connectors.py
• Lines: 1-63

Signature

HostedMCPTool(
    tool_config={
        "type": "mcp",
        "server_label": "google_calendar",
        "connector_id": "connector_googlecalendar",
        "authorization": authorization,
        "require_approval": "never",
    }
)

Import

from agents import Agent, HostedMCPTool, Runner, RunResult, RunResultStreaming

I/O Contract

Inputs

Name	Type	Required	Description
tool_config.type	str	Yes	Must be "mcp" to indicate an MCP tool
tool_config.server_label	str	Yes	Human-readable label for the MCP server/connector
tool_config.connector_id	str	Yes	The OpenAI platform connector identifier (e.g., "connector_googlecalendar")
tool_config.authorization	str	Yes	OAuth access token for the connected service
tool_config.require_approval	str	Yes	Approval policy: "never", "always", or other supported values
input	str	Yes	The user query to the agent
stream	bool	No	Whether to use streaming mode (default False)

Outputs

Name	Type	Description
run_result.final_output	str	The agent's final text response after processing MCP tool results
run_result.new_items	list	All items produced during the run, including MCP tool call results
stream events	StreamEvent	Individual streaming events including MCP events, output items, and text deltas

Usage Examples

Google Calendar Query with Streaming

import asyncio
import json
import os
from datetime import datetime
from agents import Agent, HostedMCPTool, Runner

async def main():
    authorization = os.environ["GOOGLE_CALENDAR_AUTHORIZATION"]
    agent = Agent(
        name="Assistant",
        instructions="You are a helpful assistant that can help a user with their calendar.",
        tools=[
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "google_calendar",
                    "connector_id": "connector_googlecalendar",
                    "authorization": authorization,
                    "require_approval": "never",
                }
            )
        ],
    )

    today = datetime.now().strftime("%Y-%m-%d")
    run_result = Runner.run_streamed(agent, f"What is my schedule for {today}?")
    async for event in run_result.stream_events():
        if event.type == "raw_response_event":
            if event.data.type.startswith("response.mcp"):
                print(json.dumps(event.data.to_dict(), indent=2))
            if event.data.type == "response.output_text.delta":
                print(event.data.delta, end="", flush=True)
    print()

asyncio.run(main())

Related Pages

• Environment:Openai_Openai_agents_python_OpenAI_API_Credentials