MCP tools

Ask about a small weather-station network — "which station is warmest?", "how

README Try it Source

Showcase — MCP tools

Ask about a small weather-station network — "which station is warmest?", "how does Denver's wind compare to San Francisco?" — and the model answers by calling tools on an MCP server. The server is a separate process (mcp_server.py) that speaks JSON-RPC over stdio; the backend is an MCP client that discovers the tools with tools/list, translates their JSON Schema into the model's function-calling format, and dispatches each call with tools/call. The same server drives both OpenAI and Gemini — that portability is the whole point.

Run

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Open http://localhost:3000.

To run against Gemini instead:

PROVIDER=gemini docker compose up --build

What's where

backend/mcp_server.py — the MCP server: a weather-station network exposing list_stations and get_reading. No model, no framework — just the protocol.
backend/mcp_client.py — a from-scratch MCP client: spawns the server, performs the initialize handshake, and calls tools/list / tools/call.
backend/ai_openai.py / backend/ai_gemini.py — translate MCP tools into each provider's tool format and run the bounded agent loop.
backend/main.py — identical FastAPI loader; reads PROVIDER and dispatches.
frontend/app/page.tsx — textarea + result.

The loop is capped at 8 iterations — an agent loop without a bound is an outage waiting for a confused model.

Stop

docker compose down

Run locally

Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.

Download mcp-tools.zip

unzip mcp-tools.zip
cd mcp-tools
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

Type some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.

Input

Provider

Output

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

`backend/ai_openai.py`

"""Showcase 1 (OpenAI): a model reaching an MCP server's tools.

The MCP client discovers tools over the protocol; we translate their JSON
Schema into OpenAI's function-tool format and run the usual bounded loop —
except every call is dispatched over MCP (`tools/call`), not to a local
Python function. The model never knows MCP is involved; the client does the
translation. That is the whole promise: one server, any model.
"""
import json

from openai import OpenAI

from mcp_client import MCPClient, content_text

_client = OpenAI()

_MODEL = "gpt-5.4-nano"


def _to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
    """MCP tool → OpenAI function tool. inputSchema is already JSON Schema."""
    return [{
        "type": "function",
        "name": t["name"],
        "description": t.get("description", ""),
        "parameters": t.get("inputSchema", {"type": "object", "properties": {}}),
    } for t in mcp_tools]


def run(question: str) -> str:
    with MCPClient() as mcp:
        tools = _to_openai_tools(mcp.list_tools())  # discovered at runtime
        input_list = [{"role": "user", "content": question.strip()}]
        response = None
        for _ in range(8):  # bounded — never ship an open loop
            response = _client.responses.create(
                model=_MODEL,
                instructions="Answer using the weather-station tools. Start from "
                             "list_stations if unsure of the ids. Base every number "
                             "on tool results.",
                input=input_list,
                tools=tools,
            )
            calls = [item for item in response.output if item.type == "function_call"]
            if not calls:
                break
            input_list += response.output
            for call in calls:
                args = json.loads(call.arguments)
                result = mcp.call_tool(call.name, args)  # ← dispatched over MCP
                input_list.append({
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": content_text(result),
                })
        return response.output_text if response else ""

`backend/ai_gemini.py`

"""Showcase 1 (Gemini): the same MCP server, a different model.

Nothing about the server changes. Only the translation layer differs: MCP's
JSON-Schema tools become Gemini `FunctionDeclaration`s instead of OpenAI
function tools. Point either model at the same weather-station server and it
just works — which is exactly why MCP exists.
"""
import os

from google import genai
from google.genai import types

from mcp_client import MCPClient, content_text

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

_MODEL = "gemini-3.1-flash-lite"

_TYPES = {
    "string": types.Type.STRING,
    "number": types.Type.NUMBER,
    "integer": types.Type.INTEGER,
    "boolean": types.Type.BOOLEAN,
    "array": types.Type.ARRAY,
    "object": types.Type.OBJECT,
}


def _to_gemini_schema(js: dict) -> types.Schema:
    """MCP JSON Schema → Gemini types.Schema (the fields a tool needs)."""
    js = js or {}
    t = js.get("type", "object")
    if t == "object":
        props = {k: _to_gemini_schema(v) for k, v in (js.get("properties") or {}).items()}
        return types.Schema(
            type=types.Type.OBJECT,
            properties=props,
            required=js.get("required") or None,
        )
    if t == "array":
        return types.Schema(type=types.Type.ARRAY, items=_to_gemini_schema(js.get("items") or {}))
    return types.Schema(
        type=_TYPES.get(t, types.Type.STRING),
        enum=js.get("enum"),
        description=js.get("description"),
    )


def _config(mcp_tools: list[dict]) -> types.GenerateContentConfig:
    decls = [
        types.FunctionDeclaration(
            name=t["name"],
            description=t.get("description", ""),
            parameters=_to_gemini_schema(t.get("inputSchema", {})),
        )
        for t in mcp_tools
    ]
    return types.GenerateContentConfig(
        system_instruction="Answer using the weather-station tools. Start from "
                           "list_stations if unsure of the ids. Base every number "
                           "on tool results.",
        tools=[types.Tool(function_declarations=decls)],
    )


def run(question: str) -> str:
    with MCPClient() as mcp:
        config = _config(mcp.list_tools())  # discovered at runtime
        contents = [types.Content(role="user", parts=[types.Part(text=question.strip())])]
        response = None
        for _ in range(8):  # bounded — never ship an open loop
            response = _client.models.generate_content(
                model=_MODEL, contents=contents, config=config,
            )
            if not response.function_calls:
                break
            contents.append(response.candidates[0].content)
            parts = []
            for fc in response.function_calls:
                result = mcp.call_tool(fc.name, dict(fc.args))  # ← over MCP
                parts.append(types.Part.from_function_response(
                    name=fc.name, response={"result": content_text(result)},
                ))
            contents.append(types.Content(role="user", parts=parts))
        return (response.text or "") if response else ""