Complete Example: Local Server with Agent

Updated November 2, 2026 — MCP Python SDK v2. This course now follows version 2 of the mcp SDK (2.3.0 as of October 2, 2026): MCPServer replaces FastMCP, transport settings move to run() or streamable_http_app(), and an app exposed under a real hostname must declare that hostname (transport_security). This is also the version required by the Mistral SDK: the mistralai[agents] extra requires mcp>=2.2.0,<3.0. Sources: What’s new in v2 · Migration Guide · Deploy & scale. Addendum November 3: read the result of a run_async from result.output_as_text (the mistralai SDK 3.7.0 has no output attribute), and in streaming mode the last event is a RunResult, identified with isinstance, as in the official SDK example.

Assembling the pieces

Previous lessons laid out the building blocks separately: the server and its tools, the client and its RunContext. Now you will assemble them into a working project—a local MCP server providing weather information, driven by a Mistral agent capable of answering questions in natural language. Read the code, but pay close attention to what each part decides: the server decides what is possible, the agent decides what is relevant.

The server: weather_server.py

This server consists of three tools. The first returns the weather for a city, the second compares two cities, and the third lists what the database contains. The last one is not decorative: it gives the model a way to discover the available scope instead of guessing city names.

# weather_server.py
"""Local MCP server providing weather tools."""

from mcp.server.mcpserver import MCPServer
import json

app = MCPServer("Weather Service")

# Simulated database
WEATHER_DATA = {
    "paris": {"temp": 18.5, "condition": "Cloudy", "humidity": 72},
    "lyon": {"temp": 22.0, "condition": "Sunny", "humidity": 45},
    "marseille": {"temp": 25.3, "condition": "Sunny", "humidity": 38},
    "lille": {"temp": 14.2, "condition": "Rainy", "humidity": 88},
    "bordeaux": {"temp": 20.1, "condition": "Partly cloudy", "humidity": 55},
}

@app.tool()
def get_weather(city: str) -> str:
    """Retrieves current weather for a French city.

    Args:
        city: City name in lowercase (paris, lyon, marseille, etc.)

    Returns:
        JSON with temperature, condition, and humidity.
        Returns an error if the city is unknown.
    """
    city_lower = city.lower().strip()
    if city_lower in WEATHER_DATA:
        data = WEATHER_DATA[city_lower]
        return json.dumps({
            "city": city.capitalize(),
            "temperature_celsius": data["temp"],
            "condition": data["condition"],
            "humidity_percent": data["humidity"],
        }, ensure_ascii=False)
    return json.dumps({"error": f"City {city} not found. Available cities: {', '.join(WEATHER_DATA.keys())}"})

@app.tool()
def compare_weather(city_a: str, city_b: str) -> str:
    """Compares weather between two French cities.

    Args:
        city_a: First city (lowercase)
        city_b: Second city (lowercase)

    Returns:
        Comparison of temperatures and conditions for both cities
    """
    a = WEATHER_DATA.get(city_a.lower().strip())
    b = WEATHER_DATA.get(city_b.lower().strip())
    if not a:
        return f"Unknown city: {city_a}"
    if not b:
        return f"Unknown city: {city_b}"

    diff = a["temp"] - b["temp"]
    warmer = city_a if diff > 0 else city_b
    return json.dumps({
        "city_a": {"name": city_a.capitalize(), **a},
        "city_b": {"name": city_b.capitalize(), **b},
        "temperature_difference": abs(diff),
        "warmer_city": warmer.capitalize(),
    }, ensure_ascii=False)

@app.tool()
def list_cities() -> str:
    """Lists all cities available in the weather database.

    Returns:
        List of cities with their current temperature
    """
    cities = [
        {"city": k.capitalize(), "temp": v["temp"], "condition": v["condition"]}
        for k, v in WEATHER_DATA.items()
    ]
    return json.dumps(cities, ensure_ascii=False)

Two design details deserve attention. The tools normalize input with .lower().strip() because the model might write “Paris” or “paris” depending on the user’s phrasing. And when a city is missing, get_weather does not raise an exception: it returns an error JSON that enumerates known cities, turning failure into information the model can use.

The client: main.py

The client does four things in order: instantiate Mistral, create the agent, describe how to launch the server, then open the RunContext to chain conversations.

# main.py
"""MCP client with Mistral agent for weather queries."""

import asyncio
import os
from pathlib import Path

from mistralai.client import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.mcp.stdio import MCPClientSTDIO
from mcp import StdioServerParameters
from pydantic import BaseModel

# Configuration
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
SERVER_PATH = Path(__file__).parent / "weather_server.py"

# Structured output format (optional)
class WeatherResponse(BaseModel):
    summary: str
    details: dict | None = None

async def main():
    # 1. Initialize Mistral client
    client = Mistral(api_key=MISTRAL_API_KEY)

    # 2. Create the agent
    agent = client.beta.agents.create(
        model="mistral-medium-latest",
        name="weather-assistant",
        instructions=(
            "You are a weather assistant for France. "
            "Use available tools to answer questions. "
            "Always respond in French with metric units."
        ),
    )

    # 3. Configure local MCP server
    server_params = StdioServerParameters(
        command="python",
        args=[str(SERVER_PATH)],
        env=None,
    )
    mcp_client = MCPClientSTDIO(stdio_params=server_params)

    # 4. Open context and register MCP client
    async with RunContext(
        agent_id=agent.id,
        continue_on_fn_error=True,
    ) as run_ctx:
        await run_ctx.register_mcp_client(mcp_client=mcp_client)

        # 5. Launch conversations
        questions = [
            "What's the weather like in Paris?",
            "Compare the weather between Marseille and Lille.",
            "Which city is the warmest right now?",
        ]

        for question in questions:
            print(f"\n{'='*60}")
            print(f"Question: {question}")
            print(f"{'='*60}")

            result = await client.beta.conversations.run_async(
                run_ctx=run_ctx,
                inputs=question,
            )

            print(f"Response: {result.output_as_text}")

    # 6. Automatic cleanup upon exiting context manager
    print("\nAgent and MCP server closed cleanly.")

if __name__ == "__main__":
    asyncio.run(main())

Notice the server path: Path(__file__).parent / "weather_server.py" locates the file relative to the script rather than the current directory. This is the only way to guarantee the launch works no matter where you run python main.py from.

Running the project

# Set API key
export MISTRAL_API_KEY="your_mistral_api_key"

# Launch client (it will launch the server automatically)
python main.py

You start only one process. The client creates the Mistral agent, the RunContext launches weather_server.py as an STDIO process, discovers the three tools—get_weather, compare_weather, and list_cities—then, question by question, lets the model choose which ones to call before integrating the results into its response. The third question in the example is the most interesting to observe: to know which city is the warmest, the model has no choice but to go through list_cities, even though no word in the question suggests it.

Watching the response build

Waiting for the complete response works for a script; facing a user, streaming mode offers a much better experience.

from mistralai.extra.run.result import RunResult

async with RunContext(agent_id=agent.id) as run_ctx:
    await run_ctx.register_mcp_client(mcp_client=mcp_client)

    events = await client.beta.conversations.run_stream_async(
        run_ctx=run_ctx,
        inputs="What's the weather like in Lyon?",
    )

    async for event in events:
        if isinstance(event, RunResult):
            # Last event: complete run result
            print(event.output_as_text)
        else:
            # Intermediate events (tool calls, etc.)
            print(f"[Event] {event}")

The event stream isn’t just for display: it’s also your best debugging tool, since it exposes tool calls and arguments chosen by the model in real time.

When a tool fails

Thanks to continue_on_fn_error=True, an out-of-scope request doesn’t break anything. If the user asks for the weather in Tokyo, get_weather returns its error JSON, the model reads it, and bounces back by proposing the actually available cities.

# If the user requests a non-existent city:
# → get_weather returns {"error": "City tokyo not found..."}
# → The model reads the error and reformulates its response
# → "I'm sorry, Tokyo is not in my database. Here are the available cities..."

In the end, the project fits in four files, only two of which contain logic:

weather-mcp-project/
├── weather_server.py    # MCP server (3 tools)
├── main.py              # Client with Mistral agent
├── requirements.txt     # mcp>=2.2,<3, mistralai[agents], pydantic
└── .env                 # MISTRAL_API_KEY=...

Key takeaways

• A complete MCP project = server (tools) + client (agent + RunContext)
• The server defines tools with MCPServer and @app.tool() decorators
• The client creates the agent, opens the RunContext, and launches conversations
• Streaming mode enables progressive display
• continue_on_fn_error=True handles tool errors gracefully
• All code is asynchronous—use asyncio.run() as the entry point