Complete Example: Remote Server with Auth — MCP for Developers: Servers and Integrations

Update November 2, 2026 — Le Chat is now Vibe. As of May 28, 2026, Le Chat has been rebranded as Vibe (still at chat.mistral.ai): a single agent for work and code, carrying over your conversations, settings, and subscriptions. The legacy Le Chat experience remains accessible in Chat mode; in Work mode, agents are now skills (invoked with @), memories serve as the knowledge base, and deep search is the Deep Research skill. Plans are now called Mistral Free, Pro, Team, and Enterprise. This lesson uses the name Vibe; some menus described here may have changed names or locations. Addendum November 3: run_async results are read from result.output_as_text; the mistralai SDK 3.7.0 does not have an output attribute.

Update 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.

Scenario: A Task Management MCP Server

Previous lessons covered the individual pieces: HTTP transport, middleware, and token authentication. Now you will assemble them into a fully functional project. The server exposes four task management tools—create, list, complete, delete—and the client connects remotely, presenting a bearer token with every request. This same server will then run from Vibe without a single extra line of code: that is the entire advantage of a remote server over a STDIO server.

The Server: task_server.py

Three elements coexist in this single file. FastAPI handles HTTP transport and authentication middleware, MCPServer handles the MCP protocol and tools, and a Python dictionary serves as the database. File order matters: declare the MCP server and its tools first, then create the FastAPI app with the lifespan that starts the session manager, and mount the server last. Note that the middleware protects the /tasks path, which is exactly where the MCP server is mounted: these two values must match, or you will end up with either an open door or a walled-off path.

# task_server.py
"""Remote MCP server with token authentication."""

import json
import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request, HTTPException
from mcp.server.mcpserver import MCPServer
from mcp.server.transport_security import TransportSecuritySettings

# --- Configuration ---
API_TOKEN = "mcp-secret-token-2026"

# --- In-memory database ---
tasks_db: dict[str, dict] = {}

# --- MCP Server ---
mcp = MCPServer("Task Manager")

@mcp.tool()
def create_task(title: str, description: str = "", priority: str = "medium") -> str:
    """Creates a new task in the manager.

    Args:
        title: Task title (required, max 200 characters)
        description: Detailed description (optional)
        priority: Priority level: low, medium, high (default: medium)

    Returns:
        JSON with the ID and details of the created task
    """
    task_id = str(uuid.uuid4())[:8]
    task = {
        "id": task_id,
        "title": title[:200],
        "description": description,
        "priority": priority if priority in ("low", "medium", "high") else "medium",
        "status": "open",
        "created_at": datetime.now().isoformat(),
    }
    tasks_db[task_id] = task
    return json.dumps({"success": True, "task": task}, ensure_ascii=False)

@mcp.tool()
def list_tasks(status: str = "all", priority: str = "all") -> str:
    """Lists tasks with optional filters.

    Args:
        status: Filter by status: all, open, completed (default: all)
        priority: Filter by priority: all, low, medium, high (default: all)

    Returns:
        JSON with the list of tasks and the total count
    """
    filtered = list(tasks_db.values())
    if status != "all":
        filtered = [t for t in filtered if t["status"] == status]
    if priority != "all":
        filtered = [t for t in filtered if t["priority"] == priority]

    return json.dumps({
        "total": len(filtered),
        "tasks": filtered,
    }, ensure_ascii=False)

@mcp.tool()
def complete_task(task_id: str) -> str:
    """Marks a task as completed.

    Args:
        task_id: The unique identifier of the task

    Returns:
        Confirmation of the update or an error message
    """
    if task_id not in tasks_db:
        return json.dumps({"error": f"Task {task_id} not found"})

    tasks_db[task_id]["status"] = "completed"
    tasks_db[task_id]["completed_at"] = datetime.now().isoformat()
    return json.dumps({"success": True, "task": tasks_db[task_id]}, ensure_ascii=False)

@mcp.tool()
def delete_task(task_id: str) -> str:
    """Deletes a task from the manager.

    Args:
        task_id: The unique identifier of the task to delete

    Returns:
        Confirmation of deletion or an error message
    """
    if task_id not in tasks_db:
        return json.dumps({"error": f"Task {task_id} not found"})

    deleted = tasks_db.pop(task_id)
    return json.dumps({"success": True, "deleted_task": deleted["title"]}, ensure_ascii=False)

# --- Mount the MCP server in FastAPI ---
# streamable_http_app() produces the ASGI app for the MCP server.
# Its session manager must be started by the FastAPI lifespan,
# otherwise the server responds but every tool call fails.
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with mcp.session_manager.run():
        yield

api = FastAPI(title="Task Manager MCP", lifespan=lifespan)

# --- Authentication Middleware ---
@api.middleware("http")
async def auth_middleware(request: Request, call_next):
    """Verifies the token for all /tasks routes."""
    if request.url.path.startswith("/tasks"):
        auth = request.headers.get("Authorization", "")
        if auth != f"Bearer {API_TOKEN}":
            raise HTTPException(status_code=401, detail="Invalid token")
    return await call_next(request)

# Public hostname (ngrok, production domain) declared to the MCP server
PUBLIC_HOST = os.environ.get("MCP_PUBLIC_HOST", "localhost")
securite = TransportSecuritySettings(
    allowed_hosts=[PUBLIC_HOST, f"{PUBLIC_HOST}:*", "localhost:*", "127.0.0.1:*"]
)
api.mount("/tasks", mcp.streamable_http_app(transport_security=securite))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(api, host="0.0.0.0", port=8000)

Two implementation details deserve your attention. In create_task, the title is truncated to 200 characters and the priority defaults to medium if the received value is not in the allowed list: the model sometimes proposes plausible but out-of-contract values, and the tool must remain robust without failing. In complete_task and delete_task, an unknown ID does not raise an exception but returns a JSON error message—the model can read this, understand its mistake, and rephrase, whereas a crash would prevent that.

Launch and Expose the Server

# In a first terminal, open the ngrok tunnel
ngrok http 8000
# Public URL: https://xyz789.ngrok-free.app

# In a second terminal, launch the server declaring this hostname
MCP_PUBLIC_HOST=xyz789.ngrok-free.app python task_server.py

The MCP endpoint is then located at https://xyz789.ngrok-free.app/tasks/mcp. Note the composition of this address: the public URL provided by ngrok, followed by the mount path /tasks, then the /mcp suffix added by the SDK. Confusion about this concatenation is the leading cause of connection failures.

The Client: client_tasks.py

On the client side, nothing changes compared to a local server except the MCP client type and the addition of the authentication header. The agent, however, receives instructions that describe its role in natural language—this encourages it to confirm each action rather than silently chaining calls.

# client_tasks.py
"""MCP client for the remote task server with authentication."""

import asyncio
import os
from mistralai.client import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.mcp.streamable_http import (
    MCPClientStreamableHTTP,
    StreamableHTTPServerParams,
)

API_TOKEN = "mcp-secret-token-2026"
SERVER_URL = "https://xyz789.ngrok-free.app/tasks/mcp"

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    # Create the agent with adapted instructions
    agent = client.beta.agents.create(
        model="mistral-medium-latest",
        name="task-manager",
        instructions=(
            "You are an intelligent task manager. "
            "Use the available tools to create, list, and manage "
            "the user's tasks. Respond in English. "
            "Confirm every action performed."
        ),
    )

    # Configure the remote MCP client with authentication
    mcp_client = MCPClientStreamableHTTP(
        params=StreamableHTTPServerParams(
            url=SERVER_URL,
            headers={"Authorization": f"Bearer {API_TOKEN}"},
            timeout=30,
        )
    )

    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)

        # Conversation with the manager
        conversations = [
            "Create a high priority task: Prepare the MCP presentation",
            "Create a medium priority task: Review the documentation",
            "List all ongoing tasks",
            "Mark the first task as completed",
            "Show me the remaining tasks",
        ]

        for msg in conversations:
            print(f"\n> {msg}")
            result = await client.beta.conversations.run_async(
                run_ctx=run_ctx,
                inputs=msg,
            )
            print(f"Agent: {result.output_as_text}")

asyncio.run(main())

The message sequence is deliberate: it chains two creations, a read, an update, and a re-read. This is the minimal scenario to verify that state survives from one call to the next and that the model can retrieve the ID of a task created earlier in the conversation. If the fourth instruction fails, your problem rarely lies with the server: it usually stems from the list_tasks response not making the IDs visible enough.

Test in Mistral Vibe

The same endpoint can be added directly as a custom connector, giving you a test interface without writing a client. This is also the best way to validate your docstrings: Vibe displays them as-is during discovery, so you immediately see what the model will have in front of it to decide which tool to call.

Go to the Connectors tab in Vibe
Click Add Custom Connector
Enter Name as Task Manager, URL as https://xyz789.ngrok-free.app/tasks/mcp, then choose Auth: Token and paste mcp-secret-token-2026
Click Connect

Vibe then automatically discovers the four tools, and you can converse in natural language with your own server.

Project Organization and Next Steps

task-mcp-project/
├── task_server.py       # MCP server + FastAPI + auth
├── client_tasks.py      # Python client with agent
├── requirements.txt     # fastapi, uvicorn, mcp>=2.2,<3, mistralai[agents]
└── .env                 # MISTRAL_API_KEY, API_TOKEN

This skeleton works for prototyping, but two things disqualify it for production. The plaintext token in the source code must move to .env, and the ngrok URL changes on every restart, breaking all configured connectors. ngrok remains the ideal tool for validating an idea in minutes; permanent hosting, with its Docker, cloud, and Hugging Face Spaces options, is the subject of lesson 12.

Key Takeaways

A remote server combines FastAPI (HTTP) + MCPServer (MCP) + middleware (auth), with the public hostname declared in transport_security
Token authentication is handled via the Authorization: Bearer ... header
The MCPClientStreamableHTTP client accepts custom headers for authentication
The same server is usable from both the Python SDK and Mistral Vibe
ngrok enables rapid prototyping before production deployment