Blog / AI agents

OpenAI Agents SDK with a Remote MCP Server for Web Data

11 Sept 2026Perceptlayer8 min read

To use a remote MCP server with the OpenAI Agents SDK, create an MCPServerStreamableHttp with the server URL and an Authorization header, open it with async with, and pass it to Agent(mcp_servers=[...]). This tutorial connects the SDK to Perceptlayer's server at https://api.perceptlayer.com/mcp, answers one question with a live Google search, and caps paid call_endpoint calls at three per question, so no question costs more than 3 credits.

What do you need before you start?

Python 3.10 or later, an OpenAI API key, and a Perceptlayer API key. Install the SDK with pip install openai-agents (PyPI, 2026); the code here was checked against version 0.24.0, which also installs the mcp package it uses for the CallToolResult type. Model calls are billed by OpenAI to your OpenAI account; data calls are billed in Perceptlayer credits.

Get a free API key: keys start with sk_live_, and every account gets 1,000 free credits a month after verifying an email address. Export both keys:

pip install openai-agents
export OPENAI_API_KEY=sk-...
export PERCEPTLAYER_API_KEY=sk_live_...

The MCP server needs no install. It speaks Streamable HTTP and exposes three tools: search_endpoints and get_endpoint look up the catalogue for free, and call_endpoint runs an endpoint and spends its credits only on success. The AI agents docs list the tools and the setups for other clients.

Why connect from your code and not as a hosted tool?

The SDK has two ways to reach a remote MCP server, both described in its MCP documentation (OpenAI, 2026). HostedMCPTool hands the server URL to OpenAI, and the model calls the server from OpenAI's side without a round trip to your code. MCPServerStreamableHttp runs the MCP client in your process: your code lists the tools, and every tool call passes through it.

For this tutorial the second route is the one that matters, because a budget needs a place to stand. When calls pass through your process, you can count them, refuse them and log the credits each one spent. With the hosted tool, the calls happen where you can't intercept them, and any key the server needs goes in the tool configuration you send to OpenAI. If you don't need a hard cap, the hosted tool is less code.

How do you connect the SDK to the server?

Create the server object with the URL and the header, then open it with async with, which connects on entry and closes the session on exit. Perceptlayer's server reads the key from Authorization: Bearer (or X-API-Key), so a missing header doesn't stop the agent from searching the catalogue; it only makes call_endpoint fail with invalid_api_key.

1. Put the URL and header in params. timeout is the HTTP timeout for each request, in seconds.
2. Set cache_tools_list=True. The tool list only changes when the server is deployed, so there's no need to fetch it on every run.
3. Raise client_session_timeout_seconds. In version 0.24.0 it defaults to 5 seconds, which is shorter than many data calls take. The API allows a data source up to 60 seconds before it returns timeout, so 60 matches.

from agents.mcp import MCPServerStreamableHttp

server = MCPServerStreamableHttp(
    name="perceptlayer",
    params={
        "url": "https://api.perceptlayer.com/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['PERCEPTLAYER_API_KEY']}"},
        "timeout": 30,
    },
    cache_tools_list=True,
    client_session_timeout_seconds=60,
)

Don't put the key in params of a tool call or in the URL. The server drops an api_key sent as a tool parameter, so the key never passes through the model's context. For more on header auth versus OAuth for MCP, see MCP server authentication.

How do you cap call_endpoint calls per run?

Subclass MCPServerStreamableHttp and override call_tool. Every tool call the agent makes on this server goes through that method, so it's the one place to count paid calls and refuse the ones past your limit. A refused call returns an MCP error result, which the SDK hands back to the model as the tool's output, so the agent knows to stop searching and answer.

The override also reads the credits. Each successful call_endpoint result ends with a text block such as Credits used: 1. Credits remaining: 999., so a regular expression over the result totals what the run spent.

import re
from mcp.types import CallToolResult, TextContent

class BudgetedServer(MCPServerStreamableHttp):
    """Counts call_endpoint calls and credits, and refuses calls past the cap."""

    def __init__(self, *args, max_calls=3, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_calls = max_calls
        self.calls = 0
        self.credits = 0

    async def call_tool(self, tool_name, arguments, meta=None):
        if tool_name == "call_endpoint":
            if self.calls >= self.max_calls:
                text = f"Call budget reached: {self.max_calls} call_endpoint calls this run. Answer with what you have."
                return CallToolResult(content=[TextContent(type="text", text=text)], isError=True)
            self.calls += 1
        result = await super().call_tool(tool_name, arguments, meta)
        for block in result.content:
            used = re.search(r"Credits used: (\d+)", getattr(block, "text", ""))
            if used:
                self.credits += int(used.group(1))
        return result

The count goes up before the call, so a call that fails still uses one of the three slots, even though a failed call costs no credits. That's deliberate: a cap on attempts also stops an agent that keeps retrying a bad parameter. Runner.run(..., max_turns=10) is a second, coarser limit on the whole run. For budgeting across many scheduled runs, see budgeting credits per agent run.

What does the complete agent look like?

Here's the whole script. It builds the budgeted server, creates an agent with short instructions that name the three tools, runs one question and prints the answer with the calls and credits it used. With no model argument, the agent uses the SDK's default model; pass model= to choose one.

"""Answer one question with live web data through Perceptlayer's MCP server."""
import asyncio
import os
import re
import sys

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp
from mcp.types import CallToolResult, TextContent

MCP_URL = os.environ.get("PERCEPTLAYER_MCP_URL", "https://api.perceptlayer.com/mcp")
MAX_CALLS = 3  # paid call_endpoint calls allowed per question

INSTRUCTIONS = """You answer questions with live web data.
Use search_endpoints to find an endpoint, get_endpoint to read its parameters,
then call_endpoint to run it. Lookups are free; each call_endpoint costs credits.
Make at most 3 call_endpoint calls. Cite the URLs you used."""


class BudgetedServer(MCPServerStreamableHttp):
    """Counts call_endpoint calls and credits, and refuses calls past the cap."""

    def __init__(self, *args, max_calls=MAX_CALLS, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_calls = max_calls
        self.calls = 0
        self.credits = 0

    async def call_tool(self, tool_name, arguments, meta=None):
        if tool_name == "call_endpoint":
            if self.calls >= self.max_calls:
                text = f"Call budget reached: {self.max_calls} call_endpoint calls this run. Answer with what you have."
                return CallToolResult(content=[TextContent(type="text", text=text)], isError=True)
            self.calls += 1
        result = await super().call_tool(tool_name, arguments, meta)
        for block in result.content:
            used = re.search(r"Credits used: (\d+)", getattr(block, "text", ""))
            if used:
                self.credits += int(used.group(1))
        return result


def perceptlayer_server():
    return BudgetedServer(
        name="perceptlayer",
        params={
            "url": MCP_URL,
            "headers": {"Authorization": f"Bearer {os.environ['PERCEPTLAYER_API_KEY']}"},
            "timeout": 30,
        },
        cache_tools_list=True,
        client_session_timeout_seconds=60,
    )


async def ask(question, model=None):
    async with perceptlayer_server() as server:
        agent = Agent(
            name="Web researcher",
            instructions=INSTRUCTIONS,
            mcp_servers=[server],
            **({"model": model} if model else {}),
        )
        result = await Runner.run(agent, question, max_turns=10)
        return result.final_output, server.calls, server.credits


if __name__ == "__main__":
    question = " ".join(sys.argv[1:]) or "What are the top Google results for 'mcp authorization spec'?"
    answer, calls, credits = asyncio.run(ask(question))
    print(answer)
    print(f"\ncall_endpoint calls: {calls}, credits used: {credits}")

Run it with a question:

python agent.py "What are the best-reviewed restaurants in Berlin right now?"

We haven't run this script against the live API or a real OpenAI model for this article. We ran it against a local stub MCP server with the same three tool names and canned results, driven by a scripted model that asked for five call_endpoint calls. The run made three paid calls, refused the fourth and fifth with the budget message, and reported call_endpoint calls: 3, credits used: 3.

What does the agent get back from a search?

The agent's call_endpoint result holds the endpoint's JSON, then the credits line. For google-search.search, that's Google results with organic links, local results, related questions and related searches. Here's an excerpt of a recorded response for query=best restaurants in Berlin&hl=en&safe=off, trimmed with …:

{
  "search_information": {
    "query_displayed": "best restaurants in Berlin",
    "organic_results_count": 6
  },
  "organic_results": [
    {
      "position": 2,
      "title": "Restaurants - Berlin",
      "link": "https://guide.michelin.com/us/en/berlin-region/berlin/restaurants",
      "displayed_link": "https://guide.michelin.com › berlin-region › restaurants",
      "snippet": "Berlin and surroundings: 1-48 of 79 restaurants · POTS · Rutz · GOLVET · Bandol sur mer · Nobelhart & Schmutzig · Oukan · 136 · FACIL.",
      "source": "MICHELIN Guide",
      …
    }
  ],
  "related_questions": [
    { "question": "What is the coolest restaurant in Berlin?", … }
  ],
  "local_results": {
    "places": [
      {
        "title": "BLESS Restaurant",
        "rating": 4.8,
        "reviews": 2400,
        "price": "€20–70",
        …
      }
    ]
  },
  …
}

That's a lot of text for the model to read, and you pay OpenAI for every token of it. The server cuts any result over 100,000 characters and says so, but well before that point, trimming helps. Our guide to trimming API responses for LLM context shows how to keep only the fields an answer needs, and the Search endpoint page lists the parameters, such as hl and gl, that keep results on topic.

What does one question cost?

Lookups are free and each call_endpoint to the Google Search endpoint costs 1 credit on success, so a question costs between 0 and 3 credits with this cap. Failed calls cost nothing. Most questions need one search; the cap is there for the ones where the model keeps refining its query.

Worked example: 1,000 questions a month at the cap is 1,000 × 3 × 1 credit = 3,000 credits, which fits three times in a Starter pack of 10,000 credits. If most questions use one search, the 1,000 free monthly credits cover about 1,000 of them. Purchased credits last 12 months; see pricing for what credits cost and the credits docs for how billing works.

Rate limits are per account, shared by all its keys: 1 request a second on the free tier and 5 or more on paid packs. A rate-limited call comes back to the agent as a rate_limited tool error with the seconds to wait, and it costs nothing. The rate limits docs have the numbers, and MCP server rate limiting covers how agents should handle them.

What can go wrong, and how do you fix it?

Most failures show up as tool errors the agent can read, and none of them spends credits. The table covers the ones you're most likely to see on a first run. Every error message names a code, and the errors docs list them all.

To go further, our MCP vs function calling guide compares this setup with defining the search as a plain function tool, as in adding live web search to an OpenAI agent. To run this agent now, get a free API key and try the search in the playground on the Google Search API page first, so you know what the agent will see.

openai agents sdkmcppythonai agentsweb search