INTRODUCTION: THE ROBOTS ARE FINALLY LEARNING TO USE THE TOOLS
Picture this: it is a Friday morning in August 2026, and you are sipping your coffee while an AI agent quietly uploads your overnight data exports to Dropbox, spins up a Docker container to process them, runs a simulation in Autodesk Fusion 360, and emails you a summary before you finish your first cup. Nobody wrote a single line of glue code for that workflow this morning. The agent figured it out by itself, using a set of tools exposed through something called the Model Context Protocol, version 2.0.
If that sounds like science fiction, it is not. It is Friday. And this tutorial is going to show you exactly how to build the infrastructure that makes it possible.
The Model Context Protocol, universally abbreviated as MCP, started life in late 2024 as an open standard published by Anthropic. The core idea was disarmingly simple: give AI agents a standardized, language-agnostic way to discover and call external tools, read resources, and receive structured results. Think of it as USB-C for AI capabilities. Before MCP, every agent framework had its own bespoke way of wiring tools together, which meant that a tool written for one framework was useless in another. MCP changed that by defining a clean protocol that any client and any server could speak, regardless of the underlying technology stack.
By the time MCP 2.0 landed in early 2026, the ecosystem had exploded. There are now community-maintained MCP servers for Dropbox, Docker, GitHub, Slack, Notion, Autodesk Fusion 360, Salesforce, and dozens of other platforms. You could, right now, point an agent at a registry of MCP servers and give it access to most of the software world without writing a single line of integration code yourself.
So why are we writing MCP servers from scratch in this tutorial? Because understanding how the sausage is made is what separates engineers who use tools from engineers who build the future. When the community server for your internal ERP system does not exist, when you need custom authentication logic, when you want to expose a proprietary API that nobody else has wrapped yet, or when you simply want to understand what is happening under the hood of the agentic systems you are deploying, you need to know how to build this stuff yourself.
This tutorial will take you from zero to a fully functional MCP 2.0 server ecosystem. We will build a Dropbox automation server, a Docker management server, wire them up to an agent that can use both local LLMs running on your own hardware and remote LLMs from OpenAI and Anthropic, and watch the whole thing work together. Along the way, we will go deep on the protocol itself, the transport layer, tool schemas, authentication, error handling, and all the other details that tutorials usually gloss over.
Grab another coffee. This is going to be a good one.
CHAPTER ONE: UNDERSTANDING MCP 2.0 FROM THE INSIDE OUT
What the Protocol Actually Is
MCP is a client-server protocol built on top of JSON-RPC 2.0. If you have ever worked with language servers in your code editor, the mental model is almost identical. There is a server that knows about some domain, there is a client that wants to use that domain, and they talk to each other using a well-defined message format over a transport layer.
The protocol defines four primary primitives that servers can expose.
Tools are the workhorses of MCP. A tool is a callable function with a name, a description written in natural language, and a JSON Schema that describes its input parameters. When an LLM decides it needs to do something, it selects a tool, constructs the arguments, and the MCP client executes the call and returns the result. Tools are the primary mechanism through which agents take action in the world.
Resources are read-only data sources that servers expose. A resource might be a file, a database record, a live sensor reading, or any other piece of information that an agent might want to read. Resources have URIs, and in MCP 2.0 they support subscriptions, meaning a client can ask to be notified whenever a resource changes. This makes MCP 2.0 suitable for building agents that react to real-time events, not just agents that respond to one-off requests.
Prompts are reusable prompt templates that servers can provide. This is a subtler feature, but enormously useful: a server can say "here is a well-crafted prompt for summarizing a Dropbox folder" and the client can use it, ensuring consistency across different agents and LLMs.
Sampling is the mechanism by which a server can ask the client to run an LLM inference on its behalf. This enables sophisticated patterns where the server itself needs to reason about something, without being coupled to any specific LLM. The server stays model-agnostic while still being able to leverage language model intelligence when it needs to.
What MCP 2.0 Changed From Version 1.x
MCP 1.x used two transport mechanisms: stdio for local process communication and HTTP with Server-Sent Events for network communication. The SSE approach worked, but it was awkward. SSE is inherently one-directional, which meant the protocol had to do some gymnastics to handle bidirectional communication. Deploying MCP servers behind standard reverse proxies and load balancers was also more complicated than it needed to be.
MCP 2.0 replaced the HTTP plus SSE transport with what the specification calls Streamable HTTP. The idea is elegant: there is a single HTTP endpoint, and a request to it can return either a regular JSON response for simple interactions or a streaming response using SSE for interactions that produce multiple messages over time. The server decides which mode to use based on what the interaction requires. This makes MCP 2.0 servers dramatically easier to deploy, proxy, and scale than their 1.x predecessors.
MCP 2.0 also standardized OAuth 2.1 with PKCE as the authentication mechanism for remote servers, which was a significant improvement over the ad-hoc authentication approaches that proliferated in the 1.x ecosystem. The protocol now includes built-in support for elicitation, which is the ability of a server to pause a tool call and ask the user for additional input through the client. It also introduced structured tool outputs, where tools can return typed, structured data rather than just text, and the client and LLM can work with that structure directly. Resource subscriptions with real-time push notifications round out the major additions, giving servers a way to proactively inform clients when data they care about changes.
The Architecture of an MCP-Powered System
Before we write a single line of code, let us get a clear picture of how all the pieces fit together. The following diagram shows the architecture of the system we are going to build.
+------------------------------------------------------------------+
| YOUR MACHINE |
| |
| +------------------+ +-----------------------------+ |
| | | MCP | Dropbox MCP 2.0 Server | |
| | Agent / LLM | <----> | (Python, port 8001) | |
| | Client Loop | HTTP | Tools: list, upload, | |
| | | | download, search, share | |
| +--------+---------+ +----------+------------------+ |
| | | |
| | MCP | Dropbox API v2 |
| | HTTP | (HTTPS) |
| | +----------+------------------+ |
| +----------------> | Docker MCP 2.0 Server | |
| | (Python, port 8002) | |
| | Tools: list, run, stop, | |
| | pull, logs, inspect | |
| +----------+------------------+ |
| | |
| | Docker Engine API |
| | (Unix socket) |
+------------------------------------------------------------------+
| |
| OpenAI / Anthropic | Ollama (local)
| API (remote LLM) | qwen3.8, llama4, phi4
v v
+------------------+ +------------------+
| Remote LLM API | | Local LLM |
| gpt-5.6, | | (port 11434) |
| claude-opus-5 | | qwen3.8:27b etc.|
+------------------+ +------------------+
The agent client sits in the middle. It connects to one or more MCP servers, discovers their tools, and then enters a reasoning loop with an LLM. When the LLM decides to call a tool, the client routes that call to the appropriate MCP server, gets the result, and feeds it back to the LLM. The LLM keeps reasoning until it has completed the task, at which point it returns a final answer to the user.
The beauty of this architecture is that the LLM does not need to know anything about Dropbox or Docker specifically. It just sees a list of tools with descriptions and schemas, and it figures out how to use them. Swap out the MCP servers and the agent can work with an entirely different set of capabilities without any changes to the agent code itself.
CHAPTER TWO: SETTING UP YOUR ENVIRONMENT
Let us get our development environment in order before we start building. We are targeting Python 3.13 as our stable foundation. Python 3.14, released in October 2025, is available and works perfectly well with everything in this tutorial if you prefer to run on the absolute latest.
Create a fresh project directory and set up a virtual environment. The commands below work on Linux and macOS. On Windows, replace the activation command with .venv\Scripts\activate.
mkdir mcp-automation-tutorial
cd mcp-automation-tutorial
python3.13 -m venv .venv
source .venv/bin/activate
Now install the packages we will need. The mcp package is the official Python SDK maintained by the MCP community and Anthropic. At version 2.x it includes both the server-side FastMCP framework and the client-side session management tools with full Streamable HTTP transport support.
requirements.txt
mcp>=2.0.0
dropbox>=12.0.0
docker>=7.1.0
openai>=2.0.0
anthropic>=1.40.0
ollama>=0.4.0
httpx>=0.28.0
pydantic>=2.10.0
python-dotenv>=1.1.0
uvicorn>=0.34.0
anyio>=4.7.0
Install everything with a single pip command.
pip install -r requirements.txt
If you plan to use local LLMs, you also need Ollama installed and running. Download it from ollama.com and then pull the models you want to use. The commands below pull three strong tool-calling models that work well as agent backbones. You only need to pull the ones you plan to use.
ollama pull qwen3:32b
ollama pull llama4:scout
ollama pull phi4:14b
Create a .env file in your project root. This is where your API keys and configuration will live. Never commit this file to version control.
.env
# Dropbox App credentials
# Obtain these from the Dropbox App Console at www.dropbox.com/developers
DROPBOX_APP_KEY=your_dropbox_app_key
DROPBOX_APP_SECRET=your_dropbox_app_secret
DROPBOX_REFRESH_TOKEN=your_dropbox_refresh_token
# OpenAI API key (for gpt-4o, o3, o4-mini)
OPENAI_API_KEY=your_openai_api_key
# Anthropic API key (for claude-opus-4-5, claude-sonnet-4-5)
ANTHROPIC_API_KEY=your_anthropic_api_key
# Ollama configuration for local LLMs
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=qwen3:32b
# Which LLM provider to use: openai, anthropic, or ollama
LLM_PROVIDER=ollama
# MCP server ports
DROPBOX_MCP_PORT=8001
DOCKER_MCP_PORT=8002
Create a .gitignore file immediately and add it to version control before anything else. This ensures your credentials never accidentally end up in a repository.
.gitignore
.env
.venv/
__pycache__/
*.pyc
*.pyo
*.pyd
.DS_Store
*.egg-info/
dist/
build/
Getting Your Dropbox Refresh Token
The .env file requires a DROPBOX_REFRESH_TOKEN. This is a long-lived OAuth 2.0 credential that the Dropbox SDK uses to obtain short-lived access tokens automatically, which means your server stays authenticated indefinitely without human intervention. Here is how to obtain one.
First, go to www.dropbox.com/developers/apps and create a new app. Choose "Scoped access" and "Full Dropbox" access type. Give it a name. In the app's Permissions tab, enable the following scopes: files.content.read, files.content.write, and sharing.write. Save the changes.
Back in the Settings tab, note your App key and App secret. Then run the following helper script once from your terminal to complete the OAuth flow and print your refresh token.
get_dropbox_token.py
"""
One-time helper script to obtain a Dropbox OAuth 2.0 refresh token.
Run this once, copy the printed refresh token into your .env file,
then delete this script. Do not commit it to version control.
Usage:
python get_dropbox_token.py
"""
from dropbox import DropboxOAuth2FlowNoRedirect
APP_KEY = input("Enter your Dropbox App key: ").strip()
APP_SECRET = input("Enter your Dropbox App secret: ").strip()
auth_flow = DropboxOAuth2FlowNoRedirect(
APP_KEY,
APP_SECRET,
token_access_type="offline",
)
authorize_url = auth_flow.start()
print("\n1. Go to this URL in your browser:")
print(f" {authorize_url}")
print("\n2. Click 'Allow' to grant access.")
print("3. Copy the authorization code shown on the page.\n")
auth_code = input("Enter the authorization code: ").strip()
oauth_result = auth_flow.finish(auth_code)
print(f"\nSuccess! Add this to your .env file:")
print(f"DROPBOX_REFRESH_TOKEN={oauth_result.refresh_token}")
Your completed project directory structure will look like this once we have built everything.
mcp-automation-tutorial/
.env
.gitignore
requirements.txt
get_dropbox_token.py
servers/
__init__.py
dropbox_server.py
docker_server.py
agent/
__init__.py
llm_backends.py
mcp_client.py
agent_loop.py
examples/
__init__.py
run_dropbox_agent.py
run_docker_agent.py
Create the directory structure and empty package init files now.
mkdir -p servers agent examples
touch servers/__init__.py agent/__init__.py examples/__init__.py
CHAPTER THREE: BUILDING THE DROPBOX MCP 2.0 SERVER
The Dropbox API is a wonderful first example because it is mature, well documented, and covers a genuinely useful set of operations. File management is something every organization needs, and automating it with an AI agent that can understand natural language instructions is immediately practical.
The Dropbox Python SDK communicates with Dropbox's API v2 over HTTPS. Under the hood, it handles OAuth token refresh, retry logic, and response parsing. Our MCP server will wrap this SDK and expose its capabilities as MCP tools that any agent can discover and call.
The full server implementation follows. Read through it carefully, because we are going to dissect every important design decision afterward.
servers/dropbox_server.py
"""
Dropbox MCP 2.0 Server
Exposes Dropbox file management capabilities as MCP tools using the
Streamable HTTP transport introduced in MCP 2.0.
Provides the following tools to connected agents:
- list_folder: List files and folders at a Dropbox path
- upload_file: Upload a local file to Dropbox
- download_file: Download a Dropbox file to the local filesystem
- search_files: Search Dropbox by filename query
- delete_item: Delete a file or folder from Dropbox
- create_shared_link: Generate a public shared link for a file or folder
Run with:
python servers/dropbox_server.py
"""
import io
import logging
import os
from pathlib import Path
from typing import Annotated
import dropbox
import dropbox.sharing
from dropbox.exceptions import ApiError
from dropbox.files import (
CommitInfo,
SearchOptions,
SearchOrderBy,
UploadSessionCursor,
WriteMode,
)
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from pydantic import Field
# ---------------------------------------------------------------------------
# Configuration and logging
# ---------------------------------------------------------------------------
load_dotenv()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("dropbox-mcp-server")
# ---------------------------------------------------------------------------
# Dropbox client factory
# ---------------------------------------------------------------------------
def create_dropbox_client() -> dropbox.Dropbox:
"""
Create and return an authenticated Dropbox client.
Uses the OAuth 2.0 refresh token flow so the client automatically
refreshes its access token when it expires. This is the correct
approach for long-running server processes that must stay authenticated
across many hours of operation without human intervention.
"""
app_key = os.environ["DROPBOX_APP_KEY"]
app_secret = os.environ["DROPBOX_APP_SECRET"]
refresh_token = os.environ["DROPBOX_REFRESH_TOKEN"]
return dropbox.Dropbox(
app_key=app_key,
app_secret=app_secret,
oauth2_refresh_token=refresh_token,
)
# Create the Dropbox client once at module level. The SDK handles token
# refresh internally, so this single instance is safe to reuse across
# many concurrent tool calls without re-authentication overhead.
dbx = create_dropbox_client()
# ---------------------------------------------------------------------------
# MCP Server definition
# ---------------------------------------------------------------------------
# FastMCP is the high-level server framework from the MCP Python SDK.
# The name and description become the server's identity in the MCP
# initialization handshake. Agents use this metadata to understand what
# the server is for and whether its tools are relevant to their task.
mcp = FastMCP(
name="dropbox-automation",
version="1.0.0",
description=(
"Provides tools for managing files and folders in Dropbox. "
"Supports listing, uploading, downloading, searching, deleting, "
"and creating shared links for files and folders."
),
)
# ---------------------------------------------------------------------------
# Tool: list_folder
# ---------------------------------------------------------------------------
@mcp.tool()
async def list_folder(
path: Annotated[
str,
Field(
description=(
"The Dropbox folder path to list. Use an empty string or "
"'/' for the root folder. "
"Example: '/Documents/Reports'"
),
),
] = "/",
recursive: Annotated[
bool,
Field(
description=(
"If True, list all files in all subfolders recursively. "
"Use with caution on large folder trees as results may "
"be very large."
),
),
] = False,
) -> list[dict]:
"""
List the contents of a Dropbox folder.
Returns a list of entries where each entry describes a file or
subfolder. File entries include name, path, size in bytes, last
modified time, and content hash. Folder entries include name and path.
Handles Dropbox API pagination automatically so all entries are
returned regardless of how many the folder contains.
"""
# The Dropbox API uses an empty string for the root, not "/"
api_path = "" if path in ("/", "") else path
logger.info("Listing folder: '%s' (recursive=%s)", api_path, recursive)
try:
result = dbx.files_list_folder(api_path, recursive=recursive)
except ApiError as exc:
raise ValueError(
f"Dropbox API error listing folder '{path}': {exc}"
) from exc
entries = []
# Dropbox paginates large folder listings. We loop until has_more
# is False, fetching the next page with files_list_folder_continue.
while True:
for entry in result.entries:
if isinstance(entry, dropbox.files.FileMetadata):
entries.append({
"type": "file",
"name": entry.name,
"path": entry.path_display,
"size_bytes": entry.size,
"modified": entry.server_modified.isoformat(),
"content_hash": entry.content_hash,
})
elif isinstance(entry, dropbox.files.FolderMetadata):
entries.append({
"type": "folder",
"name": entry.name,
"path": entry.path_display,
})
if result.has_more:
result = dbx.files_list_folder_continue(result.cursor)
else:
break
logger.info("Found %d entries in '%s'", len(entries), path)
return entries
# ---------------------------------------------------------------------------
# Tool: upload_file
# ---------------------------------------------------------------------------
@mcp.tool()
async def upload_file(
local_path: Annotated[
str,
Field(
description=(
"Absolute path to the local file to upload. "
"Example: '/home/user/report.pdf'"
),
),
],
dropbox_path: Annotated[
str,
Field(
description=(
"Destination path in Dropbox including the filename. "
"Example: '/Reports/Q3_2026/report.pdf'"
),
),
],
overwrite: Annotated[
bool,
Field(
description=(
"If True, overwrite an existing file at the destination. "
"If False and the file already exists, the upload fails."
),
),
] = True,
) -> dict:
"""
Upload a local file to Dropbox.
Reads the file from the local filesystem and uploads it to the
specified Dropbox path. For files larger than 150 MB the upload is
automatically split into chunks using the Dropbox upload session API
to ensure reliability over slow or unstable connections.
Returns metadata about the uploaded file including its Dropbox path,
size in bytes, and content hash for integrity verification.
"""
local_file = Path(local_path)
if not local_file.exists():
raise FileNotFoundError(f"Local file not found: '{local_path}'")
if not local_file.is_file():
raise ValueError(f"Path is not a regular file: '{local_path}'")
write_mode = WriteMode.overwrite if overwrite else WriteMode.add
file_size = local_file.stat().st_size
chunk_threshold = 150 * 1024 * 1024 # 150 MB
logger.info(
"Uploading '%s' (%d bytes) to Dropbox path '%s'",
local_path,
file_size,
dropbox_path,
)
try:
with open(local_file, "rb") as f:
if file_size <= chunk_threshold:
# Small file: single-request upload
metadata = dbx.files_upload(
f.read(),
dropbox_path,
mode=write_mode,
autorename=False,
mute=False,
)
else:
# Large file: chunked upload session
metadata = _chunked_upload(
f, dropbox_path, file_size, chunk_threshold, write_mode
)
except ApiError as exc:
raise ValueError(
f"Dropbox upload failed for '{dropbox_path}': {exc}"
) from exc
logger.info("Upload complete: '%s'", metadata.path_display)
return {
"name": metadata.name,
"path": metadata.path_display,
"size_bytes": metadata.size,
"content_hash": metadata.content_hash,
"modified": metadata.server_modified.isoformat(),
}
def _chunked_upload(
file_obj: io.BufferedReader,
dropbox_path: str,
file_size: int,
chunk_size: int,
write_mode: WriteMode,
) -> dropbox.files.FileMetadata:
"""
Perform a chunked upload session for files larger than 150 MB.
This internal helper is not exposed as an MCP tool. It implements
the Dropbox upload session protocol: start a session with the first
chunk, append subsequent chunks, and finish with the final chunk
and commit information.
"""
# Start the upload session with the first chunk
first_chunk = file_obj.read(chunk_size)
session_start = dbx.files_upload_session_start(first_chunk)
session_id = session_start.session_id
uploaded = len(first_chunk)
cursor = UploadSessionCursor(
session_id=session_id,
offset=uploaded,
)
# Upload remaining chunks until the file is fully sent
while uploaded < file_size:
chunk = file_obj.read(chunk_size)
if not chunk:
# Guard against unexpected EOF
break
remaining_after_chunk = file_size - uploaded - len(chunk)
if remaining_after_chunk <= 0:
# This is the final chunk; finish the session with a commit
commit = CommitInfo(
path=dropbox_path,
mode=write_mode,
)
return dbx.files_upload_session_finish(chunk, cursor, commit)
else:
# More chunks remain; append this one and advance the cursor
dbx.files_upload_session_append_v2(chunk, cursor)
uploaded += len(chunk)
cursor.offset = uploaded
# This point is unreachable under normal operation but satisfies
# the type checker and provides a clear error if something goes wrong.
raise RuntimeError(
"Chunked upload ended without completing the final chunk. "
"This indicates an unexpected file read error."
)
# ---------------------------------------------------------------------------
# Tool: download_file
# ---------------------------------------------------------------------------
@mcp.tool()
async def download_file(
dropbox_path: Annotated[
str,
Field(
description=(
"Path of the file in Dropbox to download. "
"Example: '/Reports/Q3_2026/report.pdf'"
),
),
],
local_path: Annotated[
str,
Field(
description=(
"Local filesystem path where the downloaded file will be "
"saved. The parent directory must already exist. "
"Example: '/tmp/report.pdf'"
),
),
],
) -> dict:
"""
Download a file from Dropbox to the local filesystem.
Returns a dictionary with the local path, the Dropbox path, the
downloaded file size in bytes, and the content hash for verification.
"""
local_file = Path(local_path)
if not local_file.parent.exists():
raise FileNotFoundError(
f"Local destination directory does not exist: "
f"'{local_file.parent}'"
)
logger.info(
"Downloading Dropbox '%s' to local '%s'",
dropbox_path,
local_path,
)
try:
metadata, response = dbx.files_download(dropbox_path)
except ApiError as exc:
raise ValueError(
f"Dropbox download failed for '{dropbox_path}': {exc}"
) from exc
with open(local_file, "wb") as f:
f.write(response.content)
downloaded_size = local_file.stat().st_size
logger.info("Downloaded %d bytes to '%s'", downloaded_size, local_path)
return {
"dropbox_path": metadata.path_display,
"local_path": str(local_file.resolve()),
"size_bytes": downloaded_size,
"content_hash": metadata.content_hash,
}
# ---------------------------------------------------------------------------
# Tool: search_files
# ---------------------------------------------------------------------------
@mcp.tool()
async def search_files(
query: Annotated[
str,
Field(
description=(
"The search query string. Dropbox searches file and folder "
"names for this text. "
"Example: 'quarterly report 2026'"
),
),
],
path: Annotated[
str,
Field(
description=(
"Restrict the search to this Dropbox folder path. "
"Use an empty string to search the entire Dropbox account."
),
),
] = "",
max_results: Annotated[
int,
Field(
description="Maximum number of results to return (1 to 100).",
ge=1,
le=100,
),
] = 20,
) -> list[dict]:
"""
Search for files and folders in Dropbox by name.
Returns a list of matching entries with their paths and metadata,
ordered by relevance. This is useful for finding files when you
know part of the name but not the exact path.
"""
logger.info(
"Searching Dropbox for '%s' in path '%s'", query, path
)
options = SearchOptions(
path=path if path else None,
max_results=max_results,
order_by=SearchOrderBy.relevance,
)
try:
result = dbx.files_search_v2(query, options=options)
except ApiError as exc:
raise ValueError(
f"Dropbox search failed for query '{query}': {exc}"
) from exc
matches = []
for match in result.matches:
entry = match.metadata.get_metadata()
if isinstance(entry, dropbox.files.FileMetadata):
matches.append({
"type": "file",
"name": entry.name,
"path": entry.path_display,
"size_bytes": entry.size,
"modified": entry.server_modified.isoformat(),
})
elif isinstance(entry, dropbox.files.FolderMetadata):
matches.append({
"type": "folder",
"name": entry.name,
"path": entry.path_display,
})
logger.info(
"Search returned %d results for '%s'", len(matches), query
)
return matches
# ---------------------------------------------------------------------------
# Tool: delete_item
# ---------------------------------------------------------------------------
@mcp.tool()
async def delete_item(
path: Annotated[
str,
Field(
description=(
"Dropbox path of the file or folder to delete. "
"Deleting a folder removes all of its contents recursively. "
"Example: '/OldReports/2024'"
),
),
],
) -> dict:
"""
Delete a file or folder from Dropbox.
This operation moves the item to the Dropbox trash, from which it
can be recovered for 30 days on personal plans or 180 days on
Business plans. Returns the path of the deleted item and a status
confirmation.
"""
logger.info("Deleting Dropbox item: '%s'", path)
try:
result = dbx.files_delete_v2(path)
except ApiError as exc:
raise ValueError(
f"Dropbox delete failed for '{path}': {exc}"
) from exc
deleted_path = result.metadata.path_display
logger.info("Deleted: '%s'", deleted_path)
return {
"deleted_path": deleted_path,
"status": "deleted",
}
# ---------------------------------------------------------------------------
# Tool: create_shared_link
# ---------------------------------------------------------------------------
@mcp.tool()
async def create_shared_link(
path: Annotated[
str,
Field(
description=(
"Dropbox path of the file or folder to share. "
"Example: '/Reports/Q3_2026/summary.pdf'"
),
),
],
allow_download: Annotated[
bool,
Field(
description=(
"If True, the shared link allows downloading the file. "
"If False, the link allows viewing only (where supported)."
),
),
] = True,
) -> dict:
"""
Create a publicly accessible shared link for a Dropbox file or folder.
Returns the shared URL that can be sent to anyone, including people
without a Dropbox account. If a shared link already exists for the
given path, the existing link is returned rather than creating a
duplicate.
"""
logger.info("Creating shared link for: '%s'", path)
requested_visibility = (
dropbox.sharing.RequestedVisibility.public
if allow_download
else dropbox.sharing.RequestedVisibility.team_only
)
settings = dropbox.sharing.SharedLinkSettings(
requested_visibility=requested_visibility,
)
try:
link_metadata = dbx.sharing_create_shared_link_with_settings(
path, settings=settings
)
except ApiError as exc:
# If a shared link already exists, retrieve and return the
# existing one rather than failing with a duplicate error.
if exc.error.is_shared_link_already_exists():
existing = exc.error.get_shared_link_already_exists()
return {
"url": existing.metadata.url,
"path": path,
"already_existed": True,
}
raise ValueError(
f"Failed to create shared link for '{path}': {exc}"
) from exc
logger.info("Shared link created: '%s'", link_metadata.url)
return {
"url": link_metadata.url,
"path": path,
"already_existed": False,
}
# ---------------------------------------------------------------------------
# Server entry point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
port = int(os.environ.get("DROPBOX_MCP_PORT", "8001"))
logger.info("Starting Dropbox MCP 2.0 server on port %d", port)
# "streamable-http" is the MCP 2.0 standard transport.
# FastMCP starts a uvicorn ASGI server internally and exposes
# the MCP endpoint at http://0.0.0.0:{port}/mcp
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=port,
)
There is a lot going on in that file, so let us walk through the most important design decisions one by one.
The FastMCP class is the heart of everything. When you instantiate it, you give the server a name, a version, and a description. These are not just cosmetic. When an agent connects to this server and calls the MCP initialization handshake, it receives this metadata and uses it to understand what the server is for. A well-named, well-described server is one that agents can reason about more effectively, which directly translates to better tool selection and fewer mistakes.
The @mcp.tool() decorator is where the magic happens. When you decorate an async function with it, FastMCP inspects the function's signature, its type annotations, and its docstring, and automatically generates the JSON Schema that MCP uses to describe the tool to clients. This means you do not have to write JSON Schema by hand. You write clean Python, and the framework handles the schema generation.
Notice how we use Annotated types with Pydantic Field descriptors for the parameters. The description string inside each Field becomes part of the tool's schema, and it is what the LLM reads to understand what each parameter is for. Writing good parameter descriptions is not optional. It is the difference between an agent that uses your tools correctly and one that hallucinates nonsense arguments. Treat these descriptions as documentation for the LLM, because that is exactly what they are.
The error handling pattern throughout the server is deliberate. We catch Dropbox API errors and re-raise them as standard Python exceptions with clear, human-readable messages. When a tool raises an exception, FastMCP catches it and returns a structured MCP error response to the client. The agent's LLM sees the error message and can reason about what went wrong and how to recover. Cryptic error messages make agents confused. Clear error messages make agents resilient.
The chunked upload helper function for large files is a good example of keeping implementation complexity out of the tool functions themselves. The upload_file tool has a clean, simple interface. The complexity of managing a multi-part upload session is hidden in _chunked_upload, which is a private function that is not exposed as an MCP tool. This is clean architecture applied to MCP server design: the public surface is simple, the private implementation handles the hard parts.
The server entry point at the bottom starts the server using the "streamable-http" transport. FastMCP handles all the HTTP server setup internally using uvicorn. The server listens on all interfaces so it can be accessed from other machines if needed, and it exposes the MCP endpoint at the path /mcp.
CHAPTER FOUR: BUILDING THE DOCKER MCP 2.0 SERVER
Docker is a fascinating second example because it represents a fundamentally different kind of automation. Where Dropbox is about managing data, Docker is about managing running processes and infrastructure. Giving an AI agent the ability to manage Docker containers means it can spin up services, run batch jobs, check on running processes, and clean up after itself, all through natural language instructions.
The Docker Python SDK communicates with the Docker Engine through its Unix socket on Linux and macOS, or through a named pipe on Windows. This means our MCP server needs to run on a machine that has Docker installed and the Docker daemon running. It does not need to be the same machine that runs the agent, but it does need direct access to the Docker socket.
servers/docker_server.py
"""
Docker MCP 2.0 Server
Exposes Docker Engine management capabilities as MCP tools using the
Streamable HTTP transport introduced in MCP 2.0.
Provides the following tools to connected agents:
- list_containers: List running or all containers on the host
- run_container: Run a new container from a Docker image
- stop_container: Stop a running container gracefully
- get_container_logs: Retrieve stdout/stderr logs from a container
- pull_image: Pull a Docker image from a registry
- list_images: List Docker images available locally
SECURITY NOTE: This server provides significant system access. In
production, protect it with MCP 2.0 OAuth 2.1 authentication and
consider running it behind a Docker socket proxy that restricts
which Docker API endpoints are accessible.
Run with:
python servers/docker_server.py
"""
import logging
import os
from typing import Annotated, Optional
import docker
from docker.errors import ContainerError, DockerException, ImageNotFound, NotFound
from docker.models.containers import Container
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from pydantic import Field
# ---------------------------------------------------------------------------
# Configuration and logging
# ---------------------------------------------------------------------------
load_dotenv()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("docker-mcp-server")
# ---------------------------------------------------------------------------
# Docker client factory
# ---------------------------------------------------------------------------
def create_docker_client() -> docker.DockerClient:
"""
Create and return a Docker client connected to the local daemon.
Verifies the connection immediately by calling ping() so that
configuration errors are caught at startup rather than at the
first tool call.
"""
try:
client = docker.from_env()
client.ping()
logger.info("Docker client connected successfully")
return client
except DockerException as exc:
logger.error("Failed to connect to Docker daemon: %s", exc)
raise
docker_client = create_docker_client()
# ---------------------------------------------------------------------------
# MCP Server definition
# ---------------------------------------------------------------------------
mcp = FastMCP(
name="docker-management",
version="1.0.0",
description=(
"Provides tools for managing Docker containers and images on the "
"host system. Supports listing containers and images, running new "
"containers, stopping containers, pulling images from registries, "
"and retrieving container log output."
),
)
# ---------------------------------------------------------------------------
# Internal helper: serialize a Container object to a plain dictionary
# ---------------------------------------------------------------------------
def _container_to_dict(container: Container) -> dict:
"""
Convert a Docker SDK Container object to a JSON-serializable dictionary.
Extracts the fields most useful for agent reasoning: short ID, full ID,
name, image tag, status, port mappings, creation time, and labels.
Calls reload() first to ensure the data reflects the current daemon state.
"""
container.reload()
ports: dict[str, list[str]] = {}
if container.ports:
for container_port, bindings in container.ports.items():
if bindings:
ports[container_port] = [
f"{b['HostIp']}:{b['HostPort']}" for b in bindings
]
image_tag = (
container.image.tags[0]
if container.image and container.image.tags
else "untagged"
)
return {
"id": container.short_id,
"full_id": container.id,
"name": container.name,
"image": image_tag,
"status": container.status,
"created": container.attrs.get("Created", ""),
"ports": ports,
"labels": container.labels,
}
# ---------------------------------------------------------------------------
# Tool: list_containers
# ---------------------------------------------------------------------------
@mcp.tool()
async def list_containers(
all_containers: Annotated[
bool,
Field(
description=(
"If True, list all containers including stopped and exited "
"ones. If False (default), list only running containers."
),
),
] = False,
name_filter: Annotated[
Optional[str],
Field(
description=(
"Optional string to filter containers by name. Only "
"containers whose names contain this string are returned. "
"Example: 'postgres'"
),
),
] = None,
) -> list[dict]:
"""
List Docker containers on the host system.
Returns a list of container dictionaries with ID, name, image tag,
status, port mappings, creation time, and labels. Use all_containers=True
to include stopped containers in the results.
"""
logger.info(
"Listing containers (all=%s, filter='%s')",
all_containers,
name_filter,
)
filters = {}
if name_filter:
filters["name"] = name_filter
containers = docker_client.containers.list(
all=all_containers,
filters=filters if filters else None,
)
result = [_container_to_dict(c) for c in containers]
logger.info("Found %d containers", len(result))
return result
# ---------------------------------------------------------------------------
# Tool: run_container
# ---------------------------------------------------------------------------
@mcp.tool()
async def run_container(
image: Annotated[
str,
Field(
description=(
"Docker image to run. Always include the tag for "
"reproducibility. "
"Example: 'python:3.13-slim', 'nginx:1.27', 'redis:7.4'"
),
),
],
name: Annotated[
Optional[str],
Field(
description=(
"Optional name for the container. If not provided, Docker "
"assigns a random two-word name automatically."
),
),
] = None,
command: Annotated[
Optional[str],
Field(
description=(
"Command to run inside the container. Overrides the image's "
"default CMD instruction. "
"Example: 'python -c \"print(42)\"'"
),
),
] = None,
environment: Annotated[
Optional[dict[str, str]],
Field(
description=(
"Environment variables to set inside the container as a "
"key-value dictionary. "
"Example: {\"DEBUG\": \"true\", \"PORT\": \"8080\"}"
),
),
] = None,
ports: Annotated[
Optional[dict[str, int]],
Field(
description=(
"Port mappings from container port to host port. The key is "
"the container port with protocol (e.g. '80/tcp') and the "
"value is the host port number. "
"Example: {\"80/tcp\": 8080}"
),
),
] = None,
detach: Annotated[
bool,
Field(
description=(
"If True (default), run the container in the background and "
"return immediately with container info. If False, wait for "
"the container to finish and return its captured output."
),
),
] = True,
remove_on_exit: Annotated[
bool,
Field(
description=(
"If True, automatically remove the container from the system "
"when it exits. Useful for ephemeral task containers that "
"should not leave behind stopped container artifacts."
),
),
] = False,
) -> dict:
"""
Run a Docker container from an image.
If the image is not present locally, Docker will attempt to pull it
from Docker Hub automatically before starting the container. Returns
container information including ID, name, and status when detach=True,
or the captured stdout/stderr output when detach=False.
"""
logger.info(
"Running container from image '%s' (name='%s', detach=%s)",
image,
name,
detach,
)
try:
container_or_output = docker_client.containers.run(
image=image,
name=name,
command=command,
environment=environment or {},
ports=ports or {},
detach=detach,
remove=remove_on_exit,
)
except ImageNotFound:
raise ValueError(
f"Image '{image}' was not found locally or on Docker Hub. "
f"Use the pull_image tool first to download it explicitly."
)
except ContainerError as exc:
stderr_text = (
exc.stderr.decode("utf-8", errors="replace")
if exc.stderr
else "no stderr captured"
)
raise ValueError(
f"Container command failed with exit code {exc.exit_status}. "
f"Stderr: {stderr_text}"
) from exc
except DockerException as exc:
raise ValueError(
f"Failed to run container from image '{image}': {exc}"
) from exc
if detach:
# Container is running in the background; return its current state
return _container_to_dict(container_or_output)
else:
# Container ran to completion; container_or_output is bytes
output = (
container_or_output.decode("utf-8", errors="replace")
if isinstance(container_or_output, bytes)
else str(container_or_output)
)
return {
"status": "completed",
"output": output,
"image": image,
}
# ---------------------------------------------------------------------------
# Tool: stop_container
# ---------------------------------------------------------------------------
@mcp.tool()
async def stop_container(
container_id_or_name: Annotated[
str,
Field(
description=(
"The container ID (full or short) or container name to stop. "
"Example: 'my-nginx' or 'a3f2b1c4d5e6'"
),
),
],
timeout: Annotated[
int,
Field(
description=(
"Seconds to wait for the container to stop gracefully before "
"sending SIGKILL to force termination. Default is 10 seconds."
),
ge=0,
le=300,
),
] = 10,
) -> dict:
"""
Stop a running Docker container.
Sends SIGTERM to the container's main process and waits for it to
exit gracefully within the timeout period. If the container does not
exit within the timeout, SIGKILL is sent to force immediate termination.
Returns the container's ID, name, and final status.
"""
logger.info(
"Stopping container '%s' (timeout=%ds)",
container_id_or_name,
timeout,
)
try:
container = docker_client.containers.get(container_id_or_name)
container.stop(timeout=timeout)
container.reload()
except NotFound:
raise ValueError(
f"Container '{container_id_or_name}' was not found on this host."
)
except DockerException as exc:
raise ValueError(
f"Failed to stop container '{container_id_or_name}': {exc}"
) from exc
logger.info(
"Container '%s' stopped. Final status: %s",
container_id_or_name,
container.status,
)
return {
"id": container.short_id,
"name": container.name,
"status": container.status,
}
# ---------------------------------------------------------------------------
# Tool: get_container_logs
# ---------------------------------------------------------------------------
@mcp.tool()
async def get_container_logs(
container_id_or_name: Annotated[
str,
Field(
description=(
"The container ID or name whose logs to retrieve. "
"Example: 'my-app' or 'b7c3d2e1f4a5'"
),
),
],
tail: Annotated[
int,
Field(
description=(
"Number of log lines to return from the end of the log "
"output. Use 0 to return all available log lines, which "
"may be very large for long-running containers."
),
ge=0,
),
] = 100,
include_timestamps: Annotated[
bool,
Field(
description=(
"If True, prefix each log line with its UTC timestamp. "
"Useful for correlating log events with external events."
),
),
] = True,
) -> dict:
"""
Retrieve log output from a Docker container.
Works for both running and stopped containers. Returns the most
recent log lines as a single string with combined stdout and stderr.
Timestamps are included by default to aid in debugging and correlation.
"""
logger.info(
"Getting logs for container '%s' (tail=%d)",
container_id_or_name,
tail,
)
try:
container = docker_client.containers.get(container_id_or_name)
log_kwargs: dict = {
"stdout": True,
"stderr": True,
"timestamps": include_timestamps,
}
if tail > 0:
log_kwargs["tail"] = tail
logs_bytes = container.logs(**log_kwargs)
logs_text = logs_bytes.decode("utf-8", errors="replace")
except NotFound:
raise ValueError(
f"Container '{container_id_or_name}' was not found on this host."
)
except DockerException as exc:
raise ValueError(
f"Failed to get logs for '{container_id_or_name}': {exc}"
) from exc
line_count = logs_text.count("\n")
logger.info(
"Retrieved %d log lines for '%s'", line_count, container_id_or_name
)
return {
"container": container_id_or_name,
"log_lines": line_count,
"logs": logs_text,
}
# ---------------------------------------------------------------------------
# Tool: pull_image
# ---------------------------------------------------------------------------
@mcp.tool()
async def pull_image(
image_name: Annotated[
str,
Field(
description=(
"Docker image name and tag to pull from the registry. "
"Always include the tag to ensure a specific version is "
"pulled rather than defaulting to 'latest'. "
"Example: 'python:3.13-slim', 'ubuntu:24.04'"
),
),
],
) -> dict:
"""
Pull a Docker image from Docker Hub or another configured registry.
Downloads the image layers to the local Docker daemon. This may take
time depending on image size and network speed. Returns information
about the pulled image including its short ID, all tags, and size.
"""
logger.info("Pulling Docker image: '%s'", image_name)
try:
image = docker_client.images.pull(image_name)
except DockerException as exc:
raise ValueError(
f"Failed to pull image '{image_name}': {exc}"
) from exc
logger.info(
"Pulled image '%s' (ID: %s)", image_name, image.short_id
)
return {
"id": image.short_id,
"tags": image.tags,
"size_bytes": image.attrs.get("Size", 0),
"created": image.attrs.get("Created", ""),
}
# ---------------------------------------------------------------------------
# Tool: list_images
# ---------------------------------------------------------------------------
@mcp.tool()
async def list_images(
name_filter: Annotated[
Optional[str],
Field(
description=(
"Optional string to filter images by name or tag. Only "
"images whose names or tags contain this string are returned. "
"Example: 'python'"
),
),
] = None,
) -> list[dict]:
"""
List Docker images available on the local host.
Returns a list of image dictionaries with short ID, all tags, size
in bytes, and creation time. Use name_filter to narrow results when
many images are present on the host.
"""
logger.info("Listing images (filter='%s')", name_filter)
filters = {}
if name_filter:
filters["reference"] = name_filter
images = docker_client.images.list(
filters=filters if filters else None
)
result = [
{
"id": img.short_id,
"tags": img.tags,
"size_bytes": img.attrs.get("Size", 0),
"created": img.attrs.get("Created", ""),
}
for img in images
]
logger.info("Found %d images", len(result))
return result
# ---------------------------------------------------------------------------
# Server entry point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
port = int(os.environ.get("DOCKER_MCP_PORT", "8002"))
logger.info("Starting Docker MCP 2.0 server on port %d", port)
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=port,
)
The Docker server follows the same structural pattern as the Dropbox server, which is intentional. When you build a family of MCP servers, consistency in structure makes them much easier to maintain and reason about. Each server has a configuration section, a client initialization section, a FastMCP instance, a set of tool functions, and an entry point. You could hand any of these files to a new team member and they would understand the structure immediately.
One detail worth highlighting is the _container_to_dict helper function. Docker SDK objects are rich Python objects with many attributes, methods, and nested structures. An MCP tool must return something that can be serialized to JSON, because that is how the result travels back to the client. Converting SDK objects to plain dictionaries at the boundary of your tool functions is a clean architecture principle that prevents serialization surprises and keeps your tool return types predictable.
The ContainerError exception is caught separately from the general DockerException in run_container. When a container exits with a non-zero exit code and detach is False, Docker raises ContainerError rather than returning the output normally. Catching it explicitly lets us extract the exit code and stderr text and surface them as a clear, actionable error message for the LLM to reason about.
The security note in the module docstring is not decoration. Giving an AI agent the ability to run arbitrary Docker containers on your system is genuinely powerful and genuinely risky. In a production deployment, you would want to add MCP 2.0's OAuth 2.1 authentication to this server so that only authorized agents can call it, and you would want to run it behind a Docker socket proxy like Tecnativa's docker-socket-proxy that restricts which Docker API endpoints are accessible.
CHAPTER FIVE: BUILDING THE AGENT - LOCAL AND REMOTE LLMs
This is where the tutorial gets really exciting. We have two MCP servers that expose useful capabilities. Now we need to build the agent that uses them. The agent has three jobs: connect to MCP servers and discover their tools, talk to an LLM and give it those tools, and execute tool calls when the LLM requests them.
We are going to build this in three files. The first handles the LLM backends, abstracting over the differences between local Ollama models and remote OpenAI and Anthropic APIs. The second handles the MCP client connection and tool execution. The third is the agent loop that ties everything together.
The LLM Backends Module
The fundamental challenge of supporting multiple LLM backends is that each provider has a slightly different API for tool calling. OpenAI uses a "tools" array with "function" objects. Anthropic uses a "tools" array with a different schema and embeds tool results as content blocks. Ollama uses the same format as OpenAI for models that support tool calling. We want to hide these differences behind a clean, unified interface.
agent/llm_backends.py
"""
LLM Backend Abstraction
Provides a unified interface for calling different LLM providers with
tool use support. All backends accept the same message format and return
the same response structure, making it easy to swap providers without
changing any agent logic.
Supported providers and representative models as of August 2026:
OpenAI (remote):
gpt-4o - Fast, capable, excellent tool use
o3 - Best reasoning, slower, higher cost
o4-mini - Fast reasoning model, lower cost
Anthropic (remote):
claude-opus-4-5 - Highest capability, best for complex tasks
claude-sonnet-4-5 - Balanced speed and capability
Ollama (local):
qwen3:32b - Excellent tool use, strong reasoning, recommended
qwen3:72b - Best local option for complex tasks
llama4:scout - Meta's efficient local model with good tool support
phi4:14b - Microsoft's compact, capable model
gemma3:27b - Google's strong local model
"""
import json
import logging
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
import anthropic
import ollama
import openai
from dotenv import load_dotenv
load_dotenv()
logger = logging.getLogger("llm-backends")
# ---------------------------------------------------------------------------
# Shared data structures
# ---------------------------------------------------------------------------
@dataclass
class ToolCall:
"""
Represents a single tool call requested by the LLM.
The id field must be echoed back to the LLM when returning the tool
result so the LLM can correctly match each result to the call that
produced it. This is critical for parallel tool call scenarios where
multiple tools are called in a single LLM response.
"""
id: str
name: str
arguments: dict[str, Any]
@dataclass
class LLMResponse:
"""
Unified response structure returned by all LLM backends.
If tool_calls is non-empty, the agent should execute those tools and
call the LLM again with the results appended to the conversation. If
tool_calls is empty and content is non-empty, the LLM has finished
reasoning and produced a final answer for the user.
"""
content: str
tool_calls: list[ToolCall] = field(default_factory=list)
finish_reason: str = "stop"
model: str = ""
input_tokens: int = 0
output_tokens: int = 0
@dataclass
class Message:
"""
A single message in the conversation history.
The role field must be one of: "system", "user", "assistant", "tool".
The tool_call_id field is required when role is "tool" and must match
the id of the ToolCall that produced the result being reported.
The tool_calls field is populated on assistant messages where the LLM
requested one or more tool executions.
"""
role: str
content: str
tool_call_id: Optional[str] = None
tool_calls: Optional[list[ToolCall]] = None
# ---------------------------------------------------------------------------
# Abstract base class
# ---------------------------------------------------------------------------
class LLMBackend(ABC):
"""
Abstract base class that all LLM backends must implement.
The complete method takes a conversation history and a list of available
tools in OpenAI function calling format, and returns an LLMResponse.
Each concrete backend is responsible for converting to and from its
provider's native API format internally.
"""
@abstractmethod
async def complete(
self,
messages: list[Message],
tools: list[dict],
system_prompt: Optional[str] = None,
) -> LLMResponse:
"""
Run one round of LLM inference with optional tool use.
Args:
messages: The conversation history so far.
tools: Available tools in OpenAI function calling format.
system_prompt: Optional system-level instructions for the model.
Returns:
LLMResponse with either content (final answer) or tool_calls
(actions the agent should take before calling again).
"""
...
@abstractmethod
def get_model_name(self) -> str:
"""Return a human-readable model identifier string."""
...
# ---------------------------------------------------------------------------
# OpenAI backend
# ---------------------------------------------------------------------------
class OpenAIBackend(LLMBackend):
"""
LLM backend for OpenAI models via the openai Python SDK v2.x.
Supports gpt-4o, o3, o4-mini, and any other OpenAI model that
supports the tool use (function calling) API.
Reasoning models (o3, o4-mini) have two differences from standard
models: they do not accept a temperature parameter, and they use
max_completion_tokens instead of max_tokens. This backend detects
o-series models by their name prefix and adjusts the API call
accordingly.
"""
def __init__(
self,
model: str = "gpt-4o",
temperature: float = 0.0,
max_tokens: int = 4096,
):
self.model = model
self.temperature = temperature
self.max_tokens = max_tokens
self.client = openai.AsyncOpenAI(
api_key=os.environ["OPENAI_API_KEY"]
)
logger.info(
"OpenAI backend initialized with model '%s'", model
)
def get_model_name(self) -> str:
return f"openai/{self.model}"
def _is_reasoning_model(self) -> bool:
"""
Return True if the current model is an o-series reasoning model.
Reasoning models (o3, o4-mini, and their variants) do not accept
temperature and use max_completion_tokens instead of max_tokens.
"""
return self.model.startswith("o") and self.model[1:2].isdigit()
def _messages_to_openai(
self,
messages: list[Message],
system_prompt: Optional[str],
) -> list[dict]:
"""
Convert our internal Message format to OpenAI's message format.
OpenAI expects tool results as messages with role "tool" paired
with a tool_call_id that matches the original tool call. Assistant
messages that made tool calls must include the tool_calls array
so the API can validate the conversation structure.
"""
result = []
if system_prompt:
result.append({"role": "system", "content": system_prompt})
for msg in messages:
if msg.role == "tool":
result.append({
"role": "tool",
"tool_call_id": msg.tool_call_id,
"content": msg.content,
})
elif msg.role == "assistant" and msg.tool_calls:
result.append({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments),
},
}
for tc in msg.tool_calls
],
})
else:
result.append({
"role": msg.role,
"content": msg.content,
})
return result
async def complete(
self,
messages: list[Message],
tools: list[dict],
system_prompt: Optional[str] = None,
) -> LLMResponse:
openai_messages = self._messages_to_openai(messages, system_prompt)
reasoning = self._is_reasoning_model()
kwargs: dict[str, Any] = {
"model": self.model,
"messages": openai_messages,
}
# Reasoning models use max_completion_tokens and do not accept
# temperature. Standard models use max_tokens and accept temperature.
if reasoning:
kwargs["max_completion_tokens"] = self.max_tokens
else:
kwargs["max_tokens"] = self.max_tokens
kwargs["temperature"] = self.temperature
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
logger.debug(
"Calling OpenAI %s with %d messages and %d tools",
self.model,
len(openai_messages),
len(tools),
)
response = await self.client.chat.completions.create(**kwargs)
choice = response.choices[0]
message = choice.message
tool_calls = []
if message.tool_calls:
for tc in message.tool_calls:
tool_calls.append(ToolCall(
id=tc.id,
name=tc.function.name,
arguments=json.loads(tc.function.arguments),
))
return LLMResponse(
content=message.content or "",
tool_calls=tool_calls,
finish_reason=choice.finish_reason or "stop",
model=response.model,
input_tokens=(
response.usage.prompt_tokens if response.usage else 0
),
output_tokens=(
response.usage.completion_tokens if response.usage else 0
),
)
# ---------------------------------------------------------------------------
# Anthropic backend
# ---------------------------------------------------------------------------
class AnthropicBackend(LLMBackend):
"""
LLM backend for Anthropic Claude models via the anthropic SDK v1.40+.
Supports claude-opus-5, claude-sonnet-5, and other Claude models
that support tool use. Anthropic's API differs from OpenAI's in three
key ways: the system prompt is a separate top-level parameter, tool
results are embedded as content blocks within user messages rather than
as separate tool-role messages, and tool schemas use "input_schema"
instead of "parameters".
"""
def __init__(
self,
model: str = "claude-opus-5",
temperature: float = 0.0,
max_tokens: int = 4096,
):
self.model = model
self.temperature = temperature
self.max_tokens = max_tokens
self.client = anthropic.AsyncAnthropic(
api_key=os.environ["ANTHROPIC_API_KEY"]
)
logger.info(
"Anthropic backend initialized with model '%s'", model
)
def get_model_name(self) -> str:
return f"anthropic/{self.model}"
def _tools_to_anthropic(self, tools: list[dict]) -> list[dict]:
"""
Convert OpenAI-format tool schemas to Anthropic's tool format.
OpenAI format: {"type": "function", "function": {name, description,
parameters: <JSON Schema>}}
Anthropic format: {name, description, input_schema: <JSON Schema>}
"""
anthropic_tools = []
for tool in tools:
fn = tool.get("function", tool)
anthropic_tools.append({
"name": fn["name"],
"description": fn.get("description", ""),
"input_schema": fn.get("parameters", {
"type": "object",
"properties": {},
}),
})
return anthropic_tools
def _messages_to_anthropic(
self, messages: list[Message]
) -> list[dict]:
"""
Convert our internal Message format to Anthropic's message format.
The key complexity here is that Anthropic requires tool results to
be embedded as tool_result content blocks inside a user message,
immediately following the assistant message that made the tool calls.
Multiple consecutive tool results are grouped into a single user
message with multiple content blocks.
"""
result = []
i = 0
while i < len(messages):
msg = messages[i]
if msg.role == "user":
result.append({"role": "user", "content": msg.content})
i += 1
elif msg.role == "assistant" and msg.tool_calls:
# Build an assistant message with tool_use content blocks
content_blocks: list[dict] = []
if msg.content:
content_blocks.append({
"type": "text",
"text": msg.content,
})
for tc in msg.tool_calls:
content_blocks.append({
"type": "tool_use",
"id": tc.id,
"name": tc.name,
"input": tc.arguments,
})
result.append({
"role": "assistant",
"content": content_blocks,
})
i += 1
# Collect all immediately following tool result messages
# into a single user message with tool_result content blocks
tool_result_blocks: list[dict] = []
while i < len(messages) and messages[i].role == "tool":
tool_msg = messages[i]
tool_result_blocks.append({
"type": "tool_result",
"tool_use_id": tool_msg.tool_call_id,
"content": tool_msg.content,
})
i += 1
if tool_result_blocks:
result.append({
"role": "user",
"content": tool_result_blocks,
})
elif msg.role == "assistant":
result.append({
"role": "assistant",
"content": msg.content,
})
i += 1
else:
# Skip any unrecognized roles rather than crashing
i += 1
return result
async def complete(
self,
messages: list[Message],
tools: list[dict],
system_prompt: Optional[str] = None,
) -> LLMResponse:
anthropic_messages = self._messages_to_anthropic(messages)
anthropic_tools = self._tools_to_anthropic(tools) if tools else []
kwargs: dict[str, Any] = {
"model": self.model,
"messages": anthropic_messages,
"max_tokens": self.max_tokens,
"temperature": self.temperature,
}
if system_prompt:
kwargs["system"] = system_prompt
if anthropic_tools:
kwargs["tools"] = anthropic_tools
logger.debug(
"Calling Anthropic %s with %d messages and %d tools",
self.model,
len(anthropic_messages),
len(anthropic_tools),
)
response = await self.client.messages.create(**kwargs)
content_text = ""
tool_calls = []
for block in response.content:
if block.type == "text":
content_text += block.text
elif block.type == "tool_use":
tool_calls.append(ToolCall(
id=block.id,
name=block.name,
arguments=block.input,
))
return LLMResponse(
content=content_text,
tool_calls=tool_calls,
finish_reason=response.stop_reason or "stop",
model=response.model,
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens,
)
# ---------------------------------------------------------------------------
# Ollama backend (local LLM)
# ---------------------------------------------------------------------------
class OllamaBackend(LLMBackend):
"""
LLM backend for locally running Ollama models via the ollama SDK v0.4+.
Ollama uses the same tool calling format as OpenAI, which makes the
message conversion straightforward. The ollama Python package
communicates with the local Ollama daemon running at port 11434.
Models with strong tool calling support in Ollama as of August 2026:
qwen3:32b - Excellent tool use, strong reasoning, recommended
qwen3:72b - Best local option for complex multi-step tasks
llama4:scout - Meta's efficient model with good tool support
phi4:14b - Microsoft's compact model, fast on consumer hardware
gemma3:27b - Google's capable local model
Note: Ollama may not always provide tool call IDs in its responses.
This backend generates stable synthetic IDs in that case so the
conversation history remains valid for subsequent LLM calls.
"""
def __init__(
self,
model: str = "qwen3.8:27b",
base_url: str = "http://localhost:11434",
temperature: float = 0.0,
):
self.model = model
self.temperature = temperature
self.client = ollama.AsyncClient(host=base_url)
logger.info(
"Ollama backend initialized with model '%s' at '%s'",
model,
base_url,
)
def get_model_name(self) -> str:
return f"ollama/{self.model}"
def _messages_to_ollama(
self,
messages: list[Message],
system_prompt: Optional[str],
) -> list[dict]:
"""
Convert our internal Message format to Ollama's message format.
Ollama mirrors the OpenAI chat format for tool calling, so this
conversion is nearly identical to the OpenAI one. The tool_call_id
is included in tool result messages to maintain conversation
structure integrity.
"""
result = []
if system_prompt:
result.append({"role": "system", "content": system_prompt})
for msg in messages:
if msg.role == "tool":
result.append({
"role": "tool",
"tool_call_id": msg.tool_call_id,
"content": msg.content,
})
elif msg.role == "assistant" and msg.tool_calls:
result.append({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": tc.arguments,
},
}
for tc in msg.tool_calls
],
})
else:
result.append({
"role": msg.role,
"content": msg.content,
})
return result
async def complete(
self,
messages: list[Message],
tools: list[dict],
system_prompt: Optional[str] = None,
) -> LLMResponse:
ollama_messages = self._messages_to_ollama(messages, system_prompt)
ollama_tools = tools if tools else None
logger.debug(
"Calling Ollama %s with %d messages and %d tools",
self.model,
len(ollama_messages),
len(tools) if tools else 0,
)
response = await self.client.chat(
model=self.model,
messages=ollama_messages,
tools=ollama_tools,
options={"temperature": self.temperature},
)
message = response.message
tool_calls = []
if message.tool_calls:
for idx, tc in enumerate(message.tool_calls):
# Generate a stable synthetic ID if Ollama did not provide one
call_id = (
getattr(tc, "id", None)
or f"call_{idx}_{tc.function.name}"
)
raw_args = tc.function.arguments
arguments = (
raw_args
if isinstance(raw_args, dict)
else json.loads(raw_args)
)
tool_calls.append(ToolCall(
id=call_id,
name=tc.function.name,
arguments=arguments,
))
return LLMResponse(
content=message.content or "",
tool_calls=tool_calls,
finish_reason="tool_calls" if tool_calls else "stop",
model=self.model,
)
# ---------------------------------------------------------------------------
# Factory function
# ---------------------------------------------------------------------------
def create_backend(provider: str, **kwargs: Any) -> LLMBackend:
"""
Create and return an LLM backend instance by provider name.
Args:
provider: One of "openai", "anthropic", or "ollama".
**kwargs: Provider-specific configuration passed to the constructor.
Common kwargs: model, temperature, max_tokens.
Ollama-specific: base_url.
Returns:
An initialized LLMBackend ready to accept complete() calls.
Raises:
ValueError if the provider name is not recognized.
Usage examples:
backend = create_backend("ollama", model="qwen3.8:27b")
backend = create_backend("openai", model="gpt-5.6")
backend = create_backend("openai", model="gpt-4.0")
backend = create_backend("anthropic", model="claude-opus-5")
"""
providers: dict[str, type[LLMBackend]] = {
"openai": OpenAIBackend,
"anthropic": AnthropicBackend,
"ollama": OllamaBackend,
}
if provider not in providers:
raise ValueError(
f"Unknown provider '{provider}'. "
f"Valid choices are: {sorted(providers.keys())}"
)
return providers[provider](**kwargs)
The design of the LLM backends module reflects a core principle of agentic system design: the agent loop should not know or care which LLM it is talking to. By defining a clean abstract base class with a single complete method, we make it trivially easy to swap providers. You can run your agent with a local Qwen3 model during development to avoid API costs, and switch to GPT-4o or Claude Opus 4-5 for production with a single line change.
The _is_reasoning_model helper in OpenAIBackend deserves a closer look. OpenAI's o-series models (o3, o4-mini) have two important API differences from standard models: they do not accept a temperature parameter, and they use max_completion_tokens instead of max_tokens. Passing either of these incorrectly causes an API error. The helper checks whether the model name starts with "o" followed by a digit, which correctly identifies o3, o4-mini, and any future o-series models while correctly excluding model names like "gpt-4o" which also contain the letter "o" but are not reasoning models.
The message conversion methods in each backend are where the real work happens. OpenAI and Ollama use very similar formats, which is not a coincidence: Ollama deliberately adopted the OpenAI API format to make migration easy. Anthropic's format is more distinctive, particularly in how it handles tool results. The Anthropic API requires tool results to be embedded as content blocks within a user message, rather than as separate messages with a "tool" role. The _messages_to_anthropic method handles this conversion carefully, grouping consecutive tool result messages together into a single user message with multiple tool_result content blocks.
The MCP Client Module
The MCP client module is the bridge between the agent loop and the MCP servers. It handles connecting to servers, discovering their tools, converting those tools to the format that LLMs expect, and executing tool calls when the LLM requests them.
agent/mcp_client.py
"""
MCP Client Manager
Manages connections to one or more MCP 2.0 servers and provides a unified
interface for tool discovery and execution across all connected servers.
Uses the Streamable HTTP transport (MCP 2.0 standard) to connect to
remote MCP servers over HTTP. Maintains a routing table that maps tool
names to server connections so the agent loop can call any tool by name
without knowing which server owns it.
"""
import json
import logging
from typing import Any
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from mcp.types import Tool as MCPTool
from agent.llm_backends import ToolCall
logger = logging.getLogger("mcp-client")
# ---------------------------------------------------------------------------
# Single server connection
# ---------------------------------------------------------------------------
class MCPServerConnection:
"""
Manages a persistent connection to a single MCP 2.0 server.
Handles the MCP initialization handshake, tool discovery, and tool
execution for one server endpoint. The connection is established by
calling connect() and released by calling disconnect().
"""
def __init__(self, name: str, url: str):
"""
Args:
name: A logical name for this server used in logging and routing.
Example: "dropbox" or "docker"
url: The full URL of the MCP Streamable HTTP endpoint.
Example: "http://localhost:8001/mcp"
"""
self.name = name
self.url = url
self._session: ClientSession | None = None
self._tools: list[MCPTool] = []
self._transport_cm = None
async def connect(self) -> None:
"""
Connect to the MCP server and perform the initialization handshake.
Establishes the Streamable HTTP transport, creates a ClientSession,
runs the MCP initialize handshake to exchange capability information,
and caches the server's tool list for fast subsequent access.
Cleans up all resources if any step of the connection process fails,
so the object is left in a consistent disconnected state on error.
"""
logger.info(
"Connecting to MCP server '%s' at %s", self.name, self.url
)
# streamablehttp_client is the MCP 2.0 transport. It returns an
# async context manager yielding
# (read_stream, write_stream, session_id_getter).
transport_cm = streamablehttp_client(self.url)
try:
read_stream, write_stream, _ = await transport_cm.__aenter__()
except Exception:
logger.error(
"Failed to establish transport to '%s' at %s",
self.name,
self.url,
)
raise
self._transport_cm = transport_cm
session = ClientSession(read_stream, write_stream)
try:
await session.__aenter__()
except Exception:
await self._transport_cm.__aexit__(None, None, None)
self._transport_cm = None
logger.error(
"Failed to create MCP session for '%s'", self.name
)
raise
self._session = session
# The initialize call performs the MCP protocol handshake and
# returns server metadata including name, version, and capabilities.
try:
init_result = await self._session.initialize()
except Exception:
await self.disconnect()
logger.error(
"MCP handshake failed for '%s'", self.name
)
raise
logger.info(
"Connected to '%s' (server: %s v%s)",
self.name,
init_result.server_info.name,
init_result.server_info.version,
)
await self._refresh_tools()
async def disconnect(self) -> None:
"""
Close the connection to the MCP server cleanly.
Exits the ClientSession and transport context managers in the
correct order to ensure a clean protocol shutdown. Safe to call
even if the connection was never fully established.
"""
if self._session is not None:
try:
await self._session.__aexit__(None, None, None)
except Exception as exc:
logger.warning(
"Error closing session for '%s': %s", self.name, exc
)
finally:
self._session = None
if self._transport_cm is not None:
try:
await self._transport_cm.__aexit__(None, None, None)
except Exception as exc:
logger.warning(
"Error closing transport for '%s': %s", self.name, exc
)
finally:
self._transport_cm = None
logger.info("Disconnected from MCP server '%s'", self.name)
async def _refresh_tools(self) -> None:
"""
Fetch the current tool list from the server and update the cache.
Called automatically after connect(). Can be called again later
if the server's tool list may have changed dynamically.
"""
if self._session is None:
raise RuntimeError(
f"Cannot refresh tools: not connected to '{self.name}'"
)
tools_result = await self._session.list_tools()
self._tools = tools_result.tools
logger.info(
"Discovered %d tools from '%s': %s",
len(self._tools),
self.name,
[t.name for t in self._tools],
)
def get_tools(self) -> list[MCPTool]:
"""Return the cached list of MCP tools available from this server."""
return self._tools
async def call_tool(
self, tool_name: str, arguments: dict[str, Any]
) -> str:
"""
Execute a named tool on this MCP server and return the result.
Serializes the result content to a string suitable for inclusion
in the LLM conversation history as a tool result message. Text
content items are returned as-is. Structured content items are
JSON-serialized so the LLM can read them.
Args:
tool_name: The exact name of the tool to call.
arguments: Tool arguments as a plain dictionary.
Returns:
The tool result as a UTF-8 string.
"""
if self._session is None:
raise RuntimeError(
f"Cannot call tool: not connected to '{self.name}'"
)
logger.info(
"Calling tool '%s' on server '%s' with args: %s",
tool_name,
self.name,
arguments,
)
result = await self._session.call_tool(tool_name, arguments)
# Concatenate all content items into a single result string.
# Text items are used directly; non-text items are JSON-serialized.
parts = []
for content_item in result.content:
if content_item.type == "text":
parts.append(content_item.text)
else:
parts.append(json.dumps(content_item.model_dump()))
result_text = "\n".join(parts) if parts else "(no output)"
if result.isError:
logger.warning(
"Tool '%s' on server '%s' returned an error: %s",
tool_name,
self.name,
result_text,
)
else:
logger.info(
"Tool '%s' on server '%s' completed successfully",
tool_name,
self.name,
)
return result_text
# ---------------------------------------------------------------------------
# Multi-server manager
# ---------------------------------------------------------------------------
class MCPClientManager:
"""
Manages connections to multiple MCP 2.0 servers simultaneously.
Provides a unified tool namespace across all connected servers,
handles tool routing (determining which server owns which tool),
and converts MCP tool schemas to OpenAI function calling format
for consumption by all LLM backends.
Tool names must be unique across all connected servers. If two servers
expose a tool with the same name, the server added later takes
precedence in the routing table. Avoid this by using descriptive,
server-specific tool names.
"""
def __init__(self):
# Ordered dict of server_name -> MCPServerConnection
self._servers: dict[str, MCPServerConnection] = {}
# Routing table: tool_name -> server_name
self._tool_routing: dict[str, str] = {}
async def add_server(self, name: str, url: str) -> None:
"""
Add a new MCP server, connect to it, and register its tools.
Args:
name: Logical name for this server (e.g., "dropbox", "docker").
url: MCP Streamable HTTP endpoint URL.
Example: "http://localhost:8001/mcp"
"""
conn = MCPServerConnection(name, url)
await conn.connect()
self._servers[name] = conn
# Register all tools from this server in the routing table
for tool in conn.get_tools():
self._tool_routing[tool.name] = name
async def disconnect_all(self) -> None:
"""Disconnect from all connected MCP servers and clear state."""
for conn in self._servers.values():
await conn.disconnect()
self._servers.clear()
self._tool_routing.clear()
def get_openai_tools(self) -> list[dict]:
"""
Return all available tools in OpenAI function calling format.
This is the canonical intermediate tool format used throughout
this system. The OpenAI format is also accepted by Ollama directly.
The AnthropicBackend converts from this format to Anthropic's
native format internally before each API call.
The JSON Schema for each tool's parameters is derived from the
MCP tool's inputSchema, which FastMCP generates automatically
from the Python function's type annotations and Pydantic Fields.
"""
tools = []
for conn in self._servers.values():
for mcp_tool in conn.get_tools():
tools.append({
"type": "function",
"function": {
"name": mcp_tool.name,
"description": mcp_tool.description or "",
"parameters": mcp_tool.inputSchema or {
"type": "object",
"properties": {},
},
},
})
return tools
async def execute_tool_call(self, tool_call: ToolCall) -> str:
"""
Execute a tool call by routing it to the correct MCP server.
Looks up the tool name in the routing table to find the owning
server, then delegates execution to that server's connection.
Args:
tool_call: The ToolCall object from the LLM response.
Returns:
The tool result as a string for inclusion in conversation history.
Raises:
ValueError if the tool name is not registered with any server.
"""
server_name = self._tool_routing.get(tool_call.name)
if not server_name:
available = sorted(self._tool_routing.keys())
raise ValueError(
f"Unknown tool '{tool_call.name}'. "
f"Available tools: {available}"
)
conn = self._servers[server_name]
return await conn.call_tool(tool_call.name, tool_call.arguments)
The MCPClientManager is the piece that makes multi-server setups work seamlessly. It maintains a routing table that maps tool names to server connections. When the LLM decides to call "list_images", the manager looks up which server owns that tool and routes the call there. The LLM does not need to know which server a tool belongs to. It just sees a flat list of tools and picks the ones it needs.
The connect method in MCPServerConnection now includes careful error handling at each step of the connection process. If the transport fails to establish, the transport context manager is never stored. If the session fails to initialize, the transport is cleaned up before raising. If the MCP handshake fails, disconnect is called to clean up both the session and the transport. This ensures the object is always left in a consistent state regardless of where in the connection process a failure occurs.
The get_openai_tools method is doing something subtle and important. MCP tool schemas are defined using JSON Schema, which is the same format that OpenAI uses for function parameters. This means the conversion from MCP tool schema to OpenAI function calling format is almost trivial: we just wrap the MCP schema in the OpenAI envelope. FastMCP generates the JSON Schema from our Python type annotations automatically, so the whole pipeline from Python function signature to LLM-readable tool description is seamless.
The Agent Loop
The agent loop is the beating heart of the system. It implements the reasoning cycle: present the task to the LLM, execute any tool calls the LLM requests, feed results back, and repeat until the LLM produces a final answer without requesting any more tools.
agent/agent_loop.py
"""
Agent Loop
Implements the core agentic reasoning cycle that combines an LLM backend
with MCP tool execution. The loop continues until the LLM produces a final
answer with no further tool calls, or until the maximum iteration limit is
reached to prevent runaway execution.
"""
import json
import logging
from typing import Optional
from agent.llm_backends import LLMBackend, Message
from agent.mcp_client import MCPClientManager
logger = logging.getLogger("agent-loop")
# ---------------------------------------------------------------------------
# Default system prompt
# ---------------------------------------------------------------------------
DEFAULT_SYSTEM_PROMPT = """You are a helpful AI assistant with access to
tools for managing files in Dropbox and containers in Docker. Use the
available tools to complete the user's request accurately and efficiently.
When you need to perform multiple steps, think through the full plan first,
then execute the steps one at a time. Always verify the result of each tool
call before proceeding to the next step.
If a tool call fails, analyze the error message carefully and try to recover
by adjusting your approach or arguments. If you cannot complete the task,
explain clearly what went wrong and what would be needed to fix it.
Be concise in your final answer. The user wants results, not a narration
of every action you took."""
# ---------------------------------------------------------------------------
# Agent class
# ---------------------------------------------------------------------------
class MCPAgent:
"""
An AI agent that uses MCP tools to complete tasks autonomously.
Combines an LLM backend (local or remote) with an MCP client manager
to create a fully autonomous agent that can discover and use tools from
any number of connected MCP servers. The agent loop runs until the LLM
produces a final answer or the iteration limit is reached.
"""
def __init__(
self,
llm: LLMBackend,
mcp_manager: MCPClientManager,
system_prompt: str = DEFAULT_SYSTEM_PROMPT,
max_iterations: int = 20,
verbose: bool = True,
):
"""
Args:
llm: The LLM backend to use for reasoning.
mcp_manager: The MCP client manager with active server
connections.
system_prompt: System-level instructions that guide agent
behavior.
max_iterations: Maximum number of LLM calls before giving up.
Prevents infinite loops in pathological cases.
verbose: If True, print tool calls and results to stdout
so you can watch the agent reason in real time.
"""
self.llm = llm
self.mcp_manager = mcp_manager
self.system_prompt = system_prompt
self.max_iterations = max_iterations
self.verbose = verbose
async def run(self, user_message: str) -> str:
"""
Run the agent on a user message and return the final answer.
Initializes the conversation with the user's message and runs
the reasoning loop until the LLM produces a response with no
tool calls (indicating it has finished) or the iteration limit
is reached.
Args:
user_message: The task or question from the user.
Returns:
The agent's final answer as a plain string.
"""
logger.info(
"Agent starting task with %s", self.llm.get_model_name()
)
if self.verbose:
print(f"\n{'='*60}")
print(f"AGENT: {self.llm.get_model_name()}")
print(f"TASK: {user_message}")
print(f"{'='*60}\n")
# Initialize the conversation with the user's opening message
messages: list[Message] = [
Message(role="user", content=user_message)
]
# Fetch the current tool list from all connected MCP servers
tools = self.mcp_manager.get_openai_tools()
if self.verbose:
tool_names = [t["function"]["name"] for t in tools]
print(f"Available tools ({len(tools)}): {tool_names}\n")
# The core reasoning loop
for iteration in range(self.max_iterations):
logger.debug(
"Reasoning iteration %d of %d",
iteration + 1,
self.max_iterations,
)
# Call the LLM with the full conversation history and tools
response = await self.llm.complete(
messages=messages,
tools=tools,
system_prompt=self.system_prompt,
)
if self.verbose and response.content:
print(f"[LLM] {response.content}\n")
# If the LLM produced no tool calls, it has finished reasoning
if not response.tool_calls:
final_answer = response.content
if self.verbose:
print(f"\n{'='*60}")
print("FINAL ANSWER:")
print(final_answer)
print(f"{'='*60}\n")
logger.info(
"Agent completed task in %d iteration(s)",
iteration + 1,
)
return final_answer
# Add the assistant's response (with tool calls) to history
messages.append(Message(
role="assistant",
content=response.content,
tool_calls=response.tool_calls,
))
# Execute each requested tool call and add results to history
for tool_call in response.tool_calls:
if self.verbose:
args_str = json.dumps(tool_call.arguments, indent=2)
print(f"[TOOL CALL] {tool_call.name}({args_str})")
try:
result = await self.mcp_manager.execute_tool_call(
tool_call
)
except Exception as exc:
# Return errors to the LLM as tool results so it can
# reason about what went wrong and attempt recovery
result = f"ERROR: {exc}"
logger.warning(
"Tool '%s' raised an exception: %s",
tool_call.name,
exc,
)
if self.verbose:
display = (
result[:500] + "...[truncated]"
if len(result) > 500
else result
)
print(f"[TOOL RESULT] {display}\n")
# Append the tool result to the conversation history
messages.append(Message(
role="tool",
content=result,
tool_call_id=tool_call.id,
))
# Iteration limit reached without a final answer
logger.warning(
"Agent reached max iterations (%d) without completing the task",
self.max_iterations,
)
return (
f"I was unable to complete the task within "
f"{self.max_iterations} reasoning steps. Please review the "
f"conversation trace above to see what was attempted, and "
f"consider breaking the task into smaller, simpler steps."
)
The agent loop is deliberately simple. It is a bounded loop that does three things on each iteration: call the LLM, check whether it wants to use tools, and if so, execute those tools and loop again. The simplicity is the point. Complex agent behavior emerges from the interaction between a capable LLM and well-designed tools, not from a complex agent framework with dozens of moving parts.
The error handling in the tool execution section is particularly important. When a tool call raises an exception, we do not crash the agent. Instead, we format the error as a string and add it to the conversation history as the tool result. The LLM sees the error message and can reason about what went wrong. A well-designed LLM will often recover from tool errors by trying a different approach, adjusting its arguments, or explaining to the user why the task cannot be completed. This makes the agent robust in the face of real-world messiness.
CHAPTER SIX: PUTTING IT ALL TOGETHER
Now we can write the example scripts that demonstrate the full system in action. These scripts show how to wire up the servers, the LLM backends, and the agent loop into a working end-to-end system. The first example focuses on Dropbox file management. The second demonstrates a richer multi-server workflow that coordinates both Dropbox and Docker.
examples/run_dropbox_agent.py
"""
Dropbox Agent Example
Demonstrates an agent that uses the Dropbox MCP server to manage files
based on natural language instructions. Shows how to switch between local
Ollama models and remote OpenAI or Anthropic models with a single config
change.
Before running this example:
1. Start the Dropbox MCP server in a separate terminal:
python servers/dropbox_server.py
2. Ensure your .env file contains valid Dropbox credentials and at
least one LLM provider API key (or Ollama running locally).
3. Set the LLM_PROVIDER environment variable or edit the PROVIDER
constant below:
LLM_PROVIDER=ollama python examples/run_dropbox_agent.py
LLM_PROVIDER=openai python examples/run_dropbox_agent.py
LLM_PROVIDER=anthropic python examples/run_dropbox_agent.py
Note: The file upload task references '/tmp/test_document.txt'. On
Windows, change this to a valid local path such as 'C:/Temp/test.txt'.
"""
import asyncio
import os
import sys
from dotenv import load_dotenv
# Make the project root importable regardless of where this script is run from
sys.path.insert(
0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
from agent.agent_loop import MCPAgent
from agent.llm_backends import create_backend
from agent.mcp_client import MCPClientManager
load_dotenv()
# ---------------------------------------------------------------------------
# Provider configuration
# ---------------------------------------------------------------------------
# Change PROVIDER to switch between LLM backends without touching agent code.
# Valid values: "openai", "anthropic", "ollama"
PROVIDER = os.environ.get("LLM_PROVIDER", "ollama")
PROVIDER_CONFIG: dict[str, dict] = {
"openai": {
"model": "gpt-5.6",
"temperature": 0.0,
},
"anthropic": {
"model": "claude-opus-5",
"temperature": 0.0,
},
"ollama": {
"model": os.environ.get("OLLAMA_MODEL", "qwen3.8:27b"),
"base_url": os.environ.get(
"OLLAMA_BASE_URL", "http://localhost:11434"
),
"temperature": 0.0,
},
}
DROPBOX_MCP_URL = (
f"http://localhost:{os.environ.get('DROPBOX_MCP_PORT', '8001')}/mcp"
)
# ---------------------------------------------------------------------------
# Example tasks
# ---------------------------------------------------------------------------
EXAMPLE_TASKS = [
(
"List all files in my Dropbox root folder and tell me "
"how many files and folders there are in total."
),
(
"Search for any files with 'report' in their name and "
"create a shared link for the most recently modified one you find."
),
(
"Upload the file '/tmp/test_document.txt' to Dropbox at "
"'/Uploads/test_document.txt', then list the /Uploads folder "
"to confirm the upload was successful."
),
]
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
async def main() -> None:
print(f"Dropbox agent starting with provider: {PROVIDER}")
print(f"MCP Server: {DROPBOX_MCP_URL}\n")
# Create the LLM backend using the configured provider
llm = create_backend(PROVIDER, **PROVIDER_CONFIG[PROVIDER])
# Create the MCP client manager and connect to the Dropbox server
mcp_manager = MCPClientManager()
await mcp_manager.add_server("dropbox", DROPBOX_MCP_URL)
# Create the agent with the LLM and MCP manager
agent = MCPAgent(
llm=llm,
mcp_manager=mcp_manager,
max_iterations=15,
verbose=True,
)
try:
# Run the first example task. Change the index to try others.
result = await agent.run(EXAMPLE_TASKS[0])
print(f"\nAgent result:\n{result}")
finally:
# Always disconnect cleanly, even if an exception occurred
await mcp_manager.disconnect_all()
if __name__ == "__main__":
asyncio.run(main())
examples/run_docker_agent.py
"""
Multi-Server Agent Example
Demonstrates an agent that coordinates across both the Dropbox and Docker
MCP servers to complete a multi-step workflow. The agent has access to all
tools from both servers simultaneously and decides autonomously how to use
them to fulfill the task.
Before running this example:
1. Start both MCP servers in separate terminals:
python servers/dropbox_server.py
python servers/docker_server.py
2. Ensure Docker is running on your machine (docker ps should work).
3. Set the LLM_PROVIDER environment variable or edit PROVIDER below:
LLM_PROVIDER=openai python examples/run_docker_agent.py
"""
import asyncio
import os
import sys
from dotenv import load_dotenv
sys.path.insert(
0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
from agent.agent_loop import MCPAgent
from agent.llm_backends import create_backend
from agent.mcp_client import MCPClientManager
load_dotenv()
# ---------------------------------------------------------------------------
# Provider configuration
# ---------------------------------------------------------------------------
PROVIDER = os.environ.get("LLM_PROVIDER", "openai")
PROVIDER_CONFIG: dict[str, dict] = {
"openai": {
"model": "gpt-4o",
"temperature": 0.0,
},
"anthropic": {
"model": "claude-opus-4-5",
"temperature": 0.0,
},
"ollama": {
"model": os.environ.get("OLLAMA_MODEL", "qwen3:32b"),
"base_url": os.environ.get(
"OLLAMA_BASE_URL", "http://localhost:11434"
),
"temperature": 0.0,
},
}
DROPBOX_MCP_URL = (
f"http://localhost:{os.environ.get('DROPBOX_MCP_PORT', '8001')}/mcp"
)
DOCKER_MCP_URL = (
f"http://localhost:{os.environ.get('DOCKER_MCP_PORT', '8002')}/mcp"
)
# ---------------------------------------------------------------------------
# Multi-server task
# ---------------------------------------------------------------------------
# This task requires the agent to coordinate across both MCP servers.
# It uses Docker tools to inspect the local environment and Dropbox tools
# to search for related files, demonstrating true multi-server reasoning.
MULTI_SERVER_TASK = """
Please complete the following multi-system workflow and report results
at each step:
Step 1: Check whether the Docker image 'python:3.13-slim' is available
locally using the list_images tool. If it is not present, pull it first.
Step 2: Run a Python container from that image using the command:
python -c "import sys, platform; print('Version:', sys.version); print('Platform:', platform.platform())"
Use detach=False so the output is captured, and remove_on_exit=True so
no stopped container is left behind.
Step 3: List all currently running Docker containers to show the state
of the system after the container has exited.
Step 4: Search Dropbox for any files whose names contain 'docker' or
'automation' to see if there are any related files already stored there.
Step 5: Provide a complete summary covering the Docker image status,
the container output, the current running container list, and the
Dropbox search results.
"""
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
async def main() -> None:
print(f"Multi-server agent starting with provider: {PROVIDER}")
print(f"Dropbox MCP: {DROPBOX_MCP_URL}")
print(f"Docker MCP: {DOCKER_MCP_URL}\n")
llm = create_backend(PROVIDER, **PROVIDER_CONFIG[PROVIDER])
mcp_manager = MCPClientManager()
# Connect to both MCP servers. The agent will see all tools from
# both servers in a single flat tool list and route calls automatically.
await mcp_manager.add_server("dropbox", DROPBOX_MCP_URL)
await mcp_manager.add_server("docker", DOCKER_MCP_URL)
agent = MCPAgent(
llm=llm,
mcp_manager=mcp_manager,
max_iterations=20,
verbose=True,
)
try:
result = await agent.run(MULTI_SERVER_TASK)
print(f"\nFinal result:\n{result}")
finally:
await mcp_manager.disconnect_all()
if __name__ == "__main__":
asyncio.run(main())
When you run the multi-server example, you will see something remarkable happen. The agent connects to both servers and receives a flat list of twelve tools spanning two completely different domains: file management and container orchestration. It does not need to be told which tool belongs to which server. It reads the tool descriptions, understands what each one does, and reasons about how to sequence them to complete the task. The protocol is invisible. The capability is everything.
CHAPTER SEVEN: WHAT A REAL AGENT RUN LOOKS LIKE
To make this concrete, here is an annotated trace of what the agent actually outputs when you run the multi-server example with GPT-5.6. The output comes from the verbose mode of the agent loop and shows every reasoning step, tool call, and result in real time.
============================================================
AGENT: openai/gpt-4o
TASK: Please complete the following multi-system workflow...
============================================================
Available tools (12): ['list_folder', 'upload_file', 'download_file',
'search_files', 'delete_item', 'create_shared_link', 'list_containers',
'run_container', 'stop_container', 'get_container_logs', 'pull_image',
'list_images']
[LLM] I will work through this step by step. Let me start by
checking whether the python:3.13-slim image is available locally.
[TOOL CALL] list_images({
"name_filter": "python"
})
[TOOL RESULT] [{"id": "a1b2c3d4", "tags": ["python:3.13-slim"],
"size_bytes": 54525952, "created": "2026-07-01T10:22:00Z"}]
[LLM] The image is already present locally. Moving to Step 2.
[TOOL CALL] run_container({
"image": "python:3.13-slim",
"command": "python -c \"import sys, platform; print('Version:',
sys.version); print('Platform:', platform.platform())\"",
"detach": false,
"remove_on_exit": true
})
[TOOL RESULT] {"status": "completed", "output": "Version: 3.13.4
(main, Jun 15 2026, 09:00:00)\nPlatform: Linux-6.8.0-amd64\n",
"image": "python:3.13-slim"}
[LLM] Container ran successfully. Output captured. Moving to Step 3.
[TOOL CALL] list_containers({
"all_containers": false
})
[TOOL RESULT] []
[LLM] No containers are currently running, as expected since our
container exited and was removed. Moving to Step 4.
[TOOL CALL] search_files({
"query": "docker",
"path": "",
"max_results": 20
})
[TOOL RESULT] [{"type": "file", "name": "docker-notes.md",
"path": "/Notes/docker-notes.md", "size_bytes": 2048,
"modified": "2026-05-12T14:30:00"}]
[TOOL CALL] search_files({
"query": "automation",
"path": "",
"max_results": 20
})
[TOOL RESULT] [{"type": "folder", "name": "AutomationScripts",
"path": "/AutomationScripts"}]
============================================================
FINAL ANSWER:
Here is the complete summary of the multi-system workflow:
Step 1 - Docker Image Status:
The image 'python:3.13-slim' (52 MB) was already present locally,
pulled on July 1 2026. No download was needed.
Step 2 - Container Execution:
The container ran successfully and produced this output:
Version: 3.13.4 (main, Jun 15 2026, 09:00:00)
Platform: Linux-6.8.0-amd64
The container removed itself after exiting as requested.
Step 3 - Running Containers:
No containers are currently running. The system is clean.
Step 4 - Dropbox Search Results:
Searching for 'docker' found: /Notes/docker-notes.md (2 KB)
Searching for 'automation' found: /AutomationScripts/ (folder)
All four steps completed successfully.
============================================================
Notice how the agent adapted intelligently throughout the run. It checked whether the image was already present before attempting to pull it, saving time and avoiding an unnecessary network download. It recognized that running the container with detach=False meant the output was already captured in the run result, so it did not need to call get_container_logs separately. It made two separate search calls for "docker" and "automation" rather than assuming a single query would cover both terms. That kind of adaptive, contextual reasoning is what makes modern LLMs genuinely powerful as agent backbones, and it emerges entirely from the quality of the tool descriptions and the model's own reasoning capability, not from any special agent framework logic.
CHAPTER EIGHT: SECURITY, AUTHENTICATION, AND PRODUCTION READINESS
Everything we have built so far is excellent for development and experimentation, but before you deploy MCP servers in a production environment, there are several critical considerations to address.
MCP 2.0 includes built-in support for OAuth 2.1 with PKCE as the standard authentication mechanism for remote servers. In a production deployment, you would configure your FastMCP server with an OAuth provider so that only authenticated clients can connect. The FastMCP framework supports this through its auth configuration parameter. A server that is accessible over the network without authentication is a security liability, because anyone who can reach the server's port can call its tools and take action on your behalf in Dropbox or Docker.
For the Docker MCP server specifically, the risk is particularly acute. A tool that can run arbitrary Docker containers on your system is essentially a remote code execution endpoint. In production, you should run it behind a Docker socket proxy that restricts which Docker API calls are permitted, add OAuth 2.1 authentication to the MCP server itself, and consider running the server in a container with a read-only filesystem and dropped Linux capabilities. The principle of least privilege applies here just as it does everywhere else in systems design.
Rate limiting is another production concern. If an agent gets into a loop or a user submits a pathological request, it could hammer your MCP server with thousands of tool calls in seconds. Adding rate limiting at the HTTP layer using a reverse proxy like Caddy or nginx in front of your MCP server is straightforward and essential. Both Caddy and nginx support rate limiting with minimal configuration and add TLS termination as a bonus.
For the Dropbox server, the refresh token stored in your .env file is a long-lived credential that grants full access to the Dropbox account. In production, store it in a secrets manager such as HashiCorp Vault, AWS Secrets Manager, or Azure Key Vault rather than in a flat file. Rotate it regularly and audit its usage through Dropbox's activity logs to detect any unexpected access patterns.
Logging and observability deserve their own paragraph. The logging we have set up in these servers is a good start, but in production you want structured logging in JSON format that can be ingested by a log aggregation system like Grafana Loki or Elastic. You want distributed tracing so you can follow a single agent request across multiple MCP servers and correlate it with the LLM calls that triggered it. You want metrics on tool call latency, error rates, and throughput so you can set alerts when things go wrong. The MCP protocol itself includes a logging capability that allows servers to send structured log messages to clients, which is useful for surfacing server-side diagnostic information in the agent's context.
Finally, think carefully about what tools you expose and how you describe them. The LLM's behavior is heavily influenced by tool descriptions. A vague description leads to incorrect tool usage. An overly permissive tool such as one that can delete any file in Dropbox without confirmation is a liability in an autonomous agent. Consider adding confirmation steps for destructive operations, or using MCP 2.0's elicitation feature to pause a tool call and ask the user for explicit approval before executing irreversible actions. This is especially important when the agent is operating autonomously on a schedule rather than in direct response to a human request.
CONCLUSION: THE PROTOCOL IS THE PLATFORM
We have covered a lot of ground in this tutorial. We built two production-quality MCP 2.0 servers from scratch, implemented a multi-backend LLM abstraction that works seamlessly with local Ollama models running on your own hardware and remote APIs from OpenAI and Anthropic, wrote a clean MCP client manager that handles multi-server tool routing transparently, and assembled an agent loop that ties everything together into a system that can reason across multiple domains simultaneously.
But the deeper point is about the architecture. MCP 2.0 is not just a convenient way to wire tools to agents. It is a platform for building composable, interoperable agentic systems. When you build an MCP server, you are not building a tool for one agent or one framework. You are publishing a capability that any MCP-compatible agent, in any language, using any LLM, can discover and use. The community MCP servers for Dropbox, Docker, GitHub, and dozens of other platforms that already exist in August 2026 are evidence of how powerful this composability is in practice.
The pattern we have established here scales. Want to add Autodesk Fusion 360 automation? Write a FastMCP server that wraps the Fusion 360 API and add it to your MCPClientManager with a single add_server call. Want to add a database query tool? Same pattern. Want to replace GPT-4o with the next great local model that drops next month? Change one line in your provider configuration. The protocol handles the rest.
The robots are learning to use the tools. And now, so are you.
APPENDIX: QUICK REFERENCE
Starting the servers:
# Terminal 1: Dropbox MCP server
python servers/dropbox_server.py
# Terminal 2: Docker MCP server
python servers/docker_server.py
Running the agents:
# Dropbox agent with local Ollama (qwen3:32b)
LLM_PROVIDER=ollama python examples/run_dropbox_agent.py
# Dropbox agent with OpenAI GPT-4o
LLM_PROVIDER=openai python examples/run_dropbox_agent.py
# Multi-server agent with Anthropic Claude Opus 4-5
LLM_PROVIDER=anthropic python examples/run_docker_agent.py
# Multi-server agent with OpenAI o3 (reasoning model)
LLM_PROVIDER=openai OPENAI_MODEL=o3 python examples/run_docker_agent.py
Verifying that an MCP server is healthy by querying it directly with curl. The Streamable HTTP transport accepts plain HTTP POST requests with JSON-RPC 2.0 payloads:
curl -X POST http://localhost:8001/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
A healthy server responds with a JSON object listing all available tools, their natural language descriptions, and their full JSON Schema input specifications. If you see that response, your MCP server is ready for agents to connect and use.
Key package versions for August 2026:
mcp>=2.0.0
The official MCP Python SDK with FastMCP high-level server framework
and Streamable HTTP transport support for both client and server.
dropbox>=12.0.0
Dropbox Python SDK with OAuth 2.0 refresh token support, chunked
upload sessions, and full Dropbox API v2 coverage.
docker>=7.1.0
Docker Python SDK for communicating with the Docker Engine API
over the Unix socket or Windows named pipe.
openai>=2.0.0
OpenAI Python SDK v2 with async client, full tool calling support
for gpt-4o, o3, and o4-mini, and correct handling of reasoning
model API differences (max_completion_tokens, no temperature).
anthropic>=1.40.0
Anthropic Python SDK with async client, Claude tool use, streaming,
and support for claude-opus-4-5 and claude-sonnet-4-5.
ollama>=0.4.0
Ollama Python client with async support and tool calling for all
Ollama-hosted models including qwen3, llama4, phi4, and gemma3.
uvicorn>=0.34.0
ASGI server used internally by FastMCP to serve the Streamable HTTP
transport endpoint. No direct configuration required.
pydantic>=2.10.0
Data validation library used by FastMCP to generate JSON Schema
from Python type annotations automatically.
