THE MAGIC BEHIND THE PIXELS: HOW AI IMAGE GENERATORS ACTUALLY WORK

My blog articles have something in common. For each article I am using DALL-E for image creation - see the image above. But how do these systems actually work? Read more …

INTRODUCTION: WHEN MATHEMATICS LEARNED TO DREAM

Imagine describing a sunset to someone who has never seen one, and having them paint it perfectly. Now imagine that “someone” is not human at all, but a mathematical function with billions of parameters that has learned to understand the relationship between words and images by studying hundreds of millions of pictures. This is not science fiction anymore. It is the remarkable reality of modern AI image generators, and understanding how they work reveals one of the most fascinating intersections of mathematics, computer science, and creativity in human history.

Just a few years ago, the idea that you could type “a steampunk octopus playing chess in a library” and receive a photorealistic image seconds later would have seemed like pure fantasy. Today, millions of people do exactly this every day using tools like DALL-E, Midjourney, and Stable Diffusion. But the journey from text prompt to stunning image involves a sophisticated dance of neural networks, probability distributions, and iterative refinement that is far more interesting than most people realize.

THE REVOLUTION NOBODY SAW COMING

The story of AI image generation is surprisingly recent. While researchers had been working on computer vision and image synthesis for decades, the breakthrough that changed everything happened between 2020 and 2022. The key innovation was not a single invention but rather the convergence of three crucial elements: a clever mathematical technique called diffusion modeling, massive datasets of images paired with text descriptions, and enough computing power to train truly enormous neural networks.

Before diffusion models took over, researchers experimented with several approaches. Generative Adversarial Networks, or GANs, dominated the field for years. These systems pit two neural networks against each other in a kind of artistic arms race. One network, the generator, tries to create fake images, while the other, the discriminator, tries to tell real images from fakes. Through this adversarial process, the generator supposedly learns to create increasingly convincing images. GANs produced some impressive results, but they were notoriously difficult to train, prone to mode collapse where they would get stuck generating similar images over and over, and struggled with generating diverse, high-quality outputs consistently.

Variational Autoencoders represented another approach, compressing images into a compact mathematical representation and then learning to reconstruct them. While elegant in theory, VAEs often produced blurry results that lacked fine detail. Researchers also explored autoregressive models that generated images pixel by pixel, treating image generation like a language modeling problem. However, this approach was computationally expensive and slow, making it impractical for high-resolution images.

Then diffusion models emerged, and everything changed.

THE DIFFUSION REVOLUTION: LEARNING TO REVERSE DESTRUCTION

At the heart of every modern AI image generator lies a surprisingly counterintuitive idea. Instead of trying to create images directly from random noise, these systems learn to gradually remove noise from corrupted images. Think of it like watching a photograph develop in reverse, starting with complete chaos and progressively revealing structure, form, and detail.

The process begins with training data consisting of millions of real images. The system takes these pristine images and systematically destroys them by adding random noise in a series of steps. Imagine taking a beautiful photograph and progressively adding more and more static until it becomes completely unrecognizable, indistinguishable from pure random noise. This destruction process is called the forward diffusion process, and it follows a precisely defined mathematical schedule. After enough steps, typically between 500 and 1000, the original image is completely obliterated.

Here is where the genius comes in. The neural network is trained to reverse this destruction. At each step of the noising process, the network learns to predict what noise was added. If the network can accurately predict the noise, it can subtract that noise to move one step backward toward the original clean image. The network never sees the task as creating an image from scratch. Instead, it learns the much more tractable problem of incremental denoising.

During training, the system shows the network images at various stages of corruption along with the noise that was added to reach that stage. The network learns patterns like “when I see this kind of partially noisy image, this specific noise pattern was likely added.” Through millions of examples, the network builds an incredibly sophisticated understanding of what constitutes noise versus signal, and how images can be progressively refined.

THE ARCHITECTURE OF IMAGINATION: INSIDE THE NEURAL NETWORK

The neural network at the core of a diffusion model is typically based on an architecture called a U-Net, which has a distinctive hourglass shape in terms of how information flows through it. The U-Net first compresses the noisy image down through several layers, extracting increasingly abstract features. At the deepest point, the network has distilled the image into a compact representation capturing its essential characteristics. Then the network expands this representation back out through decoder layers, progressively adding detail until it outputs a prediction of the noise that should be removed.

What makes this architecture so powerful is the use of skip connections that pass information directly from the encoder to the corresponding decoder layers. These shortcuts allow the network to preserve fine-grained details that might otherwise be lost in the compression process. The network can maintain both the big picture structure and the intricate textures simultaneously.

The U-Net operates on different scales through a technique called multi-scale processing. At early layers, the network works with the full resolution image, capturing fine details. As information flows deeper, the spatial resolution decreases but the network captures increasingly abstract patterns. The deepest layers understand concepts like composition, object identity, and overall scene structure, while shallower layers handle textures, edges, and local patterns. This hierarchical representation is crucial for generating coherent images with both correct high-level structure and convincing fine details.

Attention mechanisms represent another critical component. These allow different parts of the network to focus on relevant features from other parts of the image. When generating a person’s face, for instance, attention mechanisms help ensure the eyes, nose, and mouth are positioned correctly relative to each other. The network learns which spatial locations should influence each other during the generation process.

FROM WORDS TO PICTURES: THE TEXT-TO-IMAGE TRANSFORMATION

Understanding how the network handles text prompts requires diving into another layer of sophistication. The system cannot work with text directly because neural networks operate on numerical vectors, not words. The solution involves a separate neural network called a text encoder that transforms the text prompt into a rich mathematical representation.

Most modern systems use a transformer-based text encoder, often derived from language models like CLIP, which was trained to understand the relationship between images and their text descriptions. When you type a prompt like “a majestic lion standing on a cliff at sunset,” the text encoder processes these words and produces a high-dimensional vector, essentially a point in a vast mathematical space where semantically similar concepts are located near each other. This vector captures not just the individual words but their relationships and context.

The brilliance of this approach is that the text embedding space learned by CLIP has remarkable properties. Concepts that are visually or semantically related end up close together in this space. If you move through this space in a particular direction, you can smoothly transition between related concepts. The space captures visual properties, styles, and abstract relationships learned from hundreds of millions of image-text pairs.

This text embedding then guides the diffusion process through a technique called conditioning. At each denoising step, the U-Net receives both the current noisy image and the text embedding. The text embedding influences the network’s predictions, steering the denoising process toward images that match the text description. Special cross-attention layers allow the image features to attend to different parts of the text embedding, enabling fine-grained control where specific words influence specific regions of the generated image.

The conditioning mechanism works through a mathematical technique where the text embedding modulates the network’s internal activations. Think of it as the text providing a series of hints to the denoising network, saying “when you clean up this noise, steer the result toward something that looks like a lion, on a cliff, at sunset.” Each word in the prompt influences different aspects of the generation process.

THE GENERATION DANCE: FROM NOISE TO IMAGE

When you actually generate an image, the process runs in reverse compared to training. You start with pure random noise, a canvas of complete chaos where every pixel is a random value. This noise serves as the seed of creativity, the raw material from which your image will emerge. The system then runs through the denoising process step by step, guided by your text prompt.

At the first step, the noise looks completely random, but the neural network can already detect faint statistical patterns that indicate how to push this chaos toward something meaningful. The network predicts what noise should be removed to take one small step toward an image matching your prompt. That predicted noise gets subtracted, producing a slightly less noisy image.

This new, marginally cleaner image becomes the input for the next step. Again, the network predicts the noise to remove, and again, that noise is subtracted. Step by step, iteration by iteration, structure emerges from chaos. After just a few steps, you might start to see vague blobs that suggest the basic composition. More steps, and those blobs resolve into rough shapes. Continue further, and details begin to appear. Colors separate, edges sharpen, textures emerge.

The process typically requires between 20 and 100 steps, depending on the specific system and quality settings. Fewer steps generate images faster but with lower quality and less detail. More steps produce better results but take longer. The exact number represents a trade-off between speed and quality.

Each step involves running the entire U-Net forward pass, which means billions of mathematical operations. The network applies millions of learned parameters to transform the input through dozens of layers, each performing matrix multiplications, non-linear activation functions, attention computations, and normalization operations. Modern GPUs can execute these operations massively in parallel, but generating a single image still requires significant computation.

Interestingly, the process is not entirely deterministic. The random noise you start with serves as a seed for the generation. Using the same prompt but different random seeds produces different images, all matching the prompt but with variations in composition, details, and style. This stochastic nature gives the system its creative flexibility.

THE TRAINING MARATHON: HOW MODELS LEARN THEIR CRAFT

Training these models represents one of the most computationally intensive endeavors in modern machine learning. Creating a state-of-the-art image generator requires processing billions of images, each passed through the network millions of times during training. The complete training process might require thousands of GPU-hours on high-end hardware, consuming megawatts of power and costing hundreds of thousands or even millions of dollars.

The training data itself presents fascinating challenges. Modern image generators train on massive datasets scraped from the internet, typically containing hundreds of millions to billions of image-text pairs. These datasets include photographs, digital art, paintings, diagrams, and virtually every other kind of visual content humans have posted online. Each image needs associated text that describes it, which might come from surrounding webpage text, captions, alt-text, or metadata.

The quality and diversity of training data profoundly affect what the model can generate. If the dataset contains many photographs of cats but few of aardvarks, the model will generate better cats. If the dataset has extensive anime-style artwork, the model learns to generate in that style. The dataset’s biases become the model’s biases, for better or worse.

During training, the system sees the same images many times in different contexts. An image might be shown at different corruption levels, paired with different text descriptions, processed at different resolutions. This repetition with variation helps the model build robust representations that generalize beyond the specific training examples.

The optimization process uses gradient descent, where the model’s parameters are gradually adjusted to reduce prediction errors. After each batch of images, the system calculates how wrong its noise predictions were and updates billions of parameters by tiny amounts to reduce future errors. These microscopic adjustments accumulate over millions of iterations, slowly shaping the network’s behavior.

Advanced techniques like classifier-free guidance allow the model to generate images that more strongly match the text prompt by learning to predict noise both with and without text conditioning. During generation, the system can amplify the difference between these predictions, effectively turning up the strength of the text influence. This produces images that more faithfully follow prompts but sometimes at the cost of diversity and naturalness.

LATENT SPACE MAGIC: MAKING IT EFFICIENT

One innovation that made diffusion models practical was moving the diffusion process into a compressed latent space rather than operating directly on pixels. Systems like Stable Diffusion first train a separate autoencoder network that compresses images into a much smaller representation. A 512x512 pixel RGB image normally requires about 786,000 numbers to represent. The autoencoder can compress this down to a 64x64 latent representation with just a few channels, reducing the data by a factor of 48 or more.

The diffusion model then operates in this compressed latent space, learning to denoise latent representations rather than full images. This makes training and generation dramatically more efficient because the neural network processes far less data at each step. Working in latent space also has another benefit. The compressed representation naturally focuses on semantically meaningful features rather than pixel-level details, helping the model learn more abstract patterns.

After the diffusion process completes in latent space, a decoder network expands the final latent representation back into a full-resolution image. This decoder has been trained to map from the compressed representation back to detailed pixels, reconstructing fine textures and details that were abstracted away in the latent space.

The autoencoder itself must be carefully trained to preserve all information necessary to reconstruct images faithfully. If the compression loses too much detail, generated images will look blurry or distorted. If the compression is not aggressive enough, the efficiency gains disappear. Finding the right balance requires sophisticated training procedures and architectural choices.

THE COMPOSITIONAL CHALLENGE: UNDERSTANDING COMPLEX PROMPTS

One of the most impressive and mysterious capabilities of these systems is handling complex compositional prompts. When you write “a red cube on top of a blue sphere next to a green cylinder,” the model needs to understand multiple objects, their colors, their spatial relationships, and how to arrange them coherently. This requires sophisticated understanding that goes far beyond simple pattern matching.

The text encoder’s role becomes crucial here. Transformer-based encoders can capture dependencies between words, understanding that “red” modifies “cube” and “on top of” describes a spatial relationship. The attention mechanisms in the image generation network then translate these linguistic relationships into visual arrangements.

However, current models still struggle with complex compositional understanding. They might confuse which attribute applies to which object, get quantities wrong, or fail to establish correct spatial relationships. A prompt asking for “three red balls and two blue boxes” might produce two red balls and three blue boxes. Understanding why these failures occur and how to fix them remains an active area of research.

The challenge stems partly from the training data. While the dataset contains billions of images, complex multi-object compositions with specific spatial arrangements are relatively rare. The model’s understanding is built from statistical patterns in the training data, and it struggles with combinations it has seen infrequently or never seen before.

QUALITY CONTROL: WHY SOME GENERATIONS FAIL

Not every generation succeeds. Sometimes the model produces nonsensical results, anatomically impossible creatures, or images that barely relate to the prompt. Understanding why failures occur reveals the limitations of current technology.

One common issue is prompt ambiguity. If your text description could refer to multiple very different visual concepts, the model might produce something that satisfies the words but not your intent. A prompt like “Mercury” might generate the planet, the element, the Roman god, or the car brand. More specific language helps, but some ambiguity is inherent in language itself.

Another issue is the model’s statistical nature. The system learns patterns from training data but does not truly understand concepts the way humans do. It cannot reason about physics, anatomy, or causality. If asked to generate something that requires understanding these principles, like “a physically accurate depiction of a planet’s rings,” the model might produce something visually plausible but scientifically wrong.

Adversarial examples also expose the model’s brittleness. Small changes to a prompt can sometimes produce dramatically different results or cause quality to collapse. This sensitivity reveals that the model’s understanding is not as robust as human visual intelligence.

The random starting noise introduces another source of failure. Some random seeds simply lead to worse results. The optimization landscape the diffusion process navigates has many local minima, and some starting points lead to better solutions than others. This is why most tools let you generate multiple variations and cherry-pick the best ones.

THE FUTURE: WHERE IMAGE GENERATION IS HEADING

The field of AI image generation continues to evolve at breakneck speed. Several promising directions suggest where the technology might go next.

Improved control represents a major focus. Current systems offer limited ability to specify exactly what you want beyond text descriptions. Researchers are developing techniques for conditioning on additional inputs like sketches, depth maps, edge maps, or segmentation masks. These multi-modal conditioning approaches let users provide more precise guidance about composition, structure, and style.

Video generation represents the next frontier. Extending diffusion models to generate coherent video sequences poses enormous challenges because the model must maintain consistency across frames while depicting realistic motion. Early systems like Runway’s Gen-2 and OpenAI’s Sora show promise, but video generation remains far behind image generation in quality and accessibility.

Faster generation is another crucial goal. Current systems require many denoising steps, each involving expensive neural network computations. Researchers are developing distillation techniques that train smaller, faster models to mimic the outputs of larger models in fewer steps. Some recent approaches can generate high-quality images in just a single step, though usually with some quality trade-offs.

Better compositional understanding will help models handle complex prompts more reliably. This might involve new architectures specifically designed for compositional reasoning, better training procedures that emphasize compositional examples, or hybrid approaches that combine neural networks with symbolic reasoning systems.

The eventual integration with other AI systems presents exciting possibilities. Imagine describing what you want at a high level, having a language model translate that into detailed visual specifications, and having an image generator create exactly what you envisioned. Such systems would combine the reasoning capabilities of language models with the creative capabilities of image generators.

THE DEEPER IMPLICATIONS: CREATIVITY REDEFINED

Beyond the technical achievements, AI image generators raise profound questions about creativity, artistry, and the nature of visual intelligence. These systems have learned patterns from human-created art, assimilated countless styles and techniques, and gained the ability to recombine these elements in novel ways. Does this constitute genuine creativity or merely sophisticated remixing?

The debate often misses a crucial point. These systems do not simply copy or collage existing images. They build internal representations of visual concepts and relationships, then use these representations to generate completely new images that never existed before. The images are synthesized from learned patterns, not retrieved from a database. In this sense, the process shares some similarity with how human artists work, building on learned visual vocabulary and techniques while creating something new.

Yet human creativity involves aspects these systems lack. Humans create with intent, meaning, and emotional depth. We draw from lived experience, cultural context, and subjective perspective. Our art communicates ideas, tells stories, and expresses feelings. Current AI systems have none of this. They generate images that match statistical patterns but carry no intent or understanding behind them.

Perhaps the most honest assessment is that these systems represent a fundamentally new kind of visual synthesis, neither purely mechanical copying nor fully creative artistry, but something in between. They are tools that amplify human creativity, allowing people to quickly visualize ideas that would require hours or days of manual work. They democratize certain aspects of image creation while also raising legitimate concerns about the future of professional artists and the value of human-created art.

CONCLUSION: THE MAGIC CONTINUES

From random noise to photorealistic images, from mathematical abstractions to visual beauty, AI image generators represent one of the most remarkable achievements in modern computer science. The journey from a simple text prompt to a finished image involves billions of calculations, sophisticated neural architectures, and clever training procedures that teach machines to understand the relationship between language and vision.

These systems work not by magic but by mathematics, not by copying but by learning patterns, not by understanding but by statistical inference. Yet the results often appear magical because they bridge the gap between human language and visual imagination in ways that were impossible just a few years ago.

Understanding how these systems actually work reveals both their impressive capabilities and their fundamental limitations. They are powerful tools for visual synthesis, built on clever ideas about learning to reverse a destruction process, operating in compressed representation spaces, and conditioning generation on text embeddings. They represent a significant step toward machines that can understand and generate visual content, though they remain far from human visual intelligence.

As the technology continues to evolve, these systems will become faster, more capable, and more integrated into creative workflows. The mathematical foundations laid by diffusion models will likely spawn new generations of even more powerful systems. The revolution in AI-generated imagery has only just begun, and understanding the mechanics behind it helps us appreciate both what has been achieved and what remains possible.

The pixels on your screen that form a perfect sunset over an impossible landscape, a photorealistic rendering of something that never existed, a creative vision brought to life in seconds, all of this emerges from the elegant mathematics of diffusion models learning to sculpt signal from noise. That is the true magic behind the pixels.​​​​​​​​​​​​​​​​

BUILDING AN MCP SERVER FOR A NON-REST SDK

CHAPTER 1: THE BIG PICTURE - WHY MCP EXISTS AND WHY YOU SHOULD CARE

Imagine you have just hired a brilliant new assistant. This assistant can reason, plan, write, summarize, and debate philosophy. There is only one problem: they are locked in a room with no windows, no telephone, and no connection to the outside world. They can only work with whatever you slide under the door. That, in a nutshell, is what a large language model looks like without tool integration. It is extraordinarily smart, but it is also completely cut off from the living, breathing systems your organization actually runs on.

For years, developers tried to solve this problem with ad-hoc approaches. They would write custom prompt-engineering scaffolds, inject data into the system prompt by hand, build one-off integrations for each model provider, and then repeat the entire exercise when they switched from one model to another. The result was a sprawling tangle of bespoke glue code that was hard to maintain, impossible to reuse, and painful to test.

The Model Context Protocol, known universally as MCP, was Anthropic's answer to this chaos. First introduced in late 2024 and now governed by the 2026-07-28 specification, MCP is an open standard that defines a clean, provider-agnostic contract between an AI model and the external world. It answers a deceptively simple question: how should an AI system discover and use the capabilities of an external service, regardless of what language that service is written in, what transport it uses, or which AI model is on the other end of the wire?

The answer MCP gives is elegant. It defines three kinds of things an external service can offer: tools that the model can call to perform actions, resources that provide read-only data as context, and prompts that are reusable instruction templates. It then defines a wire protocol for discovering and invoking these things, and it provides official SDKs in Python, TypeScript, Go, and C# that make implementing both sides of this protocol straightforward.

The 2026-07-28 specification made one particularly important architectural decision: it went fully stateless. Earlier versions of MCP required a handshake to establish a session, complete with session identifiers and sticky routing. The new specification throws all of that away. Every request now carries everything it needs to be processed independently. This makes MCP servers trivially deployable as standard web services behind any load balancer, which is exactly where you want them in a production environment.

This tutorial will take you from zero to a fully working MCP server that wraps a real, non-REST Python SDK, and then show you how to connect both a local LLM running through Ollama and a remote LLM running through the Anthropic API to that server. By the time you finish reading, you will understand not just the mechanics of MCP but the reasoning behind every design choice.

CHAPTER 2: UNDERSTANDING THE MCP ARCHITECTURE

Before writing a single line of code, it is worth building a clear mental model of how MCP fits together. The protocol defines three roles: the host, the client, and the server.

The host is the application that the end user actually interacts with. This might be Claude Desktop, a custom chat interface you built, a CI/CD pipeline, or an autonomous agent loop. The host is responsible for managing the user experience and for deciding, at a high level, what AI capabilities to expose.

The client lives inside the host. It speaks the MCP protocol and maintains the relationship with one or more MCP servers. When the model decides it needs to call a tool, the client is the thing that actually sends that request to the right server and brings the result back.

The server is what you are going to build in this tutorial. It is an independent process or service that exposes capabilities through the MCP protocol. It knows nothing about the model, nothing about the user interface, and nothing about the host application. It simply says: here are the tools I offer, here are the resources I can provide, here are the prompts I understand. When asked to execute a tool, it executes it and returns the result.

The flow of a typical interaction looks like this:

  User

   |

   | (natural language request)

   v

  Host Application

   |

   | (sends messages to model)

   v

  LLM (Claude Sonnet 5 / local Ollama model)

   |

   | (decides to call a tool, emits tool_use block)

   v

  MCP Client (inside the host)

   |

   | (MCP protocol request over HTTP to /mcp endpoint)

   v

  MCP Server  <----> External SDK / API / Library

   |

   | (MCP protocol response)

   v

  MCP Client

   |

   | (tool_result appended to conversation)

   v

  LLM (generates final response)

   |

   v

  Host Application --> User

The MCP server in this diagram is the translation layer. Its job is to take the structured requests that arrive over the MCP protocol and translate them into calls against whatever underlying system you are wrapping, then translate the results back into structured data that the protocol understands.

The 2026-07-28 specification uses HTTP as its primary transport, specifically the Streamable HTTP transport. The official Python SDK v2 provides a Client class that handles all protocol machinery automatically: JSON-RPC framing, capability negotiation, error handling, and response parsing. You simply connect the Client to the server URL and call methods like list_tools(), call_tool(), and read_resource(). All the wire-format details are handled for you.

The server side is equally well-abstracted. The Python SDK v2 provides FastMCP, which you decorate your Python functions with using @mcp.tool(), @mcp.resource(), or @mcp.prompt(). FastMCP automatically generates the JSON schemas for the parameters, validates incoming requests, routes them to the right function, and serializes the responses. The amount of boilerplate you have to write is genuinely minimal.

CHAPTER 3: THE PERFECT EXAMPLE - PSUTIL AS A NON-REST SDK

Most MCP tutorials wrap a REST API. They pick something like a weather service or a GitHub API, show you how to make HTTP requests, and call it a day. That is fine as far as it goes, but it misses an important and very common real-world scenario: wrapping a library or SDK that is not REST-based at all.

The psutil library is a perfect illustration of this scenario. It is a cross-platform Python library that provides access to system information and process management. It does not expose an HTTP endpoint. It does not use JSON. It is a pure Python API that makes direct operating system calls through a combination of C extensions and platform-specific code. You import it, you call its functions, and you get back Python objects.

psutil is also an excellent example because its API naturally divides into exactly the two categories that are most important to understand when building an MCP server: methods that retrieve information and methods that perform actions.

On the information side, psutil gives you functions like cpu_percent(), which returns the current CPU utilization as a floating-point number, and virtual_memory(), which returns a named tuple describing RAM usage with fields like total, available, used, and percent. There is also disk_usage(), which takes a path and returns disk space statistics, cpu_count(), which tells you how many logical processors the system has, and getloadavg(), which returns the system load average over the past one, five, and fifteen minutes. These are pure queries. They observe the system but change nothing about it.

On the action side, psutil gives you functions like process termination through Process.terminate() and Process.kill(), the ability to suspend and resume processes with Process.suspend() and Process.resume(), and the ability to change process priorities with Process.nice(). These are operations that actually modify the state of the running system. They have side effects.

This distinction matters enormously when you are designing an MCP server, because MCP treats these two kinds of operations very differently. Information-retrieval operations map naturally to MCP resources, which are read-only data endpoints. Action operations map naturally to MCP tools, which are executable functions that may have side effects. Understanding this mapping is the central intellectual exercise of building any MCP server.

Here is a conceptual map of how the psutil API maps to MCP primitives:

  psutil API                     MCP Primitive     Reason

  -----------------------------------------------------------------------

  cpu_percent()                  Resource          Read-only observation

  virtual_memory()               Resource          Read-only observation

  disk_usage(path)               Resource          Read-only observation

  cpu_count()                    Resource          Read-only observation

  getloadavg()                   Resource          Read-only observation

  disk_partitions()              Resource          Read-only observation

  Process.terminate(pid)         Tool              Has side effects

  Process.kill(pid)              Tool              Has side effects

  Process.suspend(pid)           Tool              Has side effects

  Process.resume(pid)            Tool              Has side effects

  Process.nice(pid, value)       Tool              Has side effects

  process_iter()                 Tool              Parameterized query

Notice that process_iter() lands in the Tool category even though it is conceptually a query. The reason is that it takes parameters, such as what attributes to retrieve and optional filtering criteria, and in MCP, parameterized operations that require input from the model are more naturally expressed as tools than as resources. Resources are better suited to fixed-URI data endpoints that the host application can fetch proactively to build context.

This is a nuance worth dwelling on. Resources in MCP are identified by URIs, like system://cpu or system://memory. The host application can fetch these URIs to provide ambient context to the model before the conversation even starts. Tools, on the other hand, are invoked by the model on demand, based on what the model decides it needs in order to answer a question. Both are valuable, but they serve different purposes in the overall architecture.

CHAPTER 4: SETTING UP YOUR ENVIRONMENT

The project you are going to build has a clear directory structure that separates concerns cleanly. The MCP server lives in its own package, the client code for local and remote LLMs lives in separate modules, and configuration is centralized. The two __init__.py files are empty; they simply mark their directories as Python packages.

  system-monitor-mcp/

  |

  |-- server/

  |   |-- __init__.py            (empty - marks as Python package)

  |   |-- server.py              (the MCP server)

  |   |-- system_info.py         (psutil wrapper / domain layer)

  |

  |-- client/

  |   |-- __init__.py            (empty - marks as Python package)

  |   |-- client_anthropic.py    (client using Claude Sonnet 5)

  |   |-- client_ollama.py       (client using local Ollama model)

  |   |-- mcp_client.py          (shared async MCP communication layer)

  |

  |-- config.py                  (centralized configuration)

  |-- requirements.txt

  |-- run_server.py              (entry point for the server)

  |-- .env                       (API keys - never commit to version control)

  |-- Dockerfile

The requirements.txt file captures every dependency the project needs. The mcp package is the official Python SDK v2, which includes both the server-side FastMCP and the client-side Client class. The fastapi and uvicorn packages are needed to host the MCP server as an HTTP service. The psutil package is the library we are wrapping. The anthropic package is the official Anthropic Python SDK for talking to Claude Sonnet 5. The ollama package is the official Python wrapper for local Ollama models. The python-dotenv package loads environment variables from the .env file.

# requirements.txt

#

# Pin to v2 of the MCP SDK, which implements the 2026-07-28 specification.

# The v1 SDK used a stateful session model and is not compatible with this code.

mcp>=2.0.0

fastapi>=0.115.0

uvicorn[standard]>=0.32.0

psutil>=6.1.0

anthropic>=1.0.0

ollama>=0.4.0

python-dotenv>=1.0.0

Install everything in a fresh virtual environment by running these commands from the project root:

python -m venv .venv

source .venv/bin/activate

pip install -r requirements.txt

On Windows, activate the virtual environment with .venv\Scripts\activate instead.

You also need a running Ollama instance for the local LLM client. Ollama is a tool that lets you run large language models locally on your own hardware. If you do not have it installed, visit ollama.com and follow the installation instructions for your platform. Once installed, pull a model that supports tool calling:

ollama pull llama3.3

Llama 3.3 is an excellent choice because it has strong tool-calling capabilities, runs well on consumer hardware with 16 GB of RAM or more, and produces high-quality responses for system administration tasks.

For the remote LLM client, you need an Anthropic API key. Create a .env file in the project root with the following content:

ANTHROPIC_API_KEY=sk-ant-your-key-here

MCP_API_KEY=your-server-api-key-here

The python-dotenv library will automatically load this file when any module calls load_dotenv(). Never commit the .env file to version control. Add it to your .gitignore.

CHAPTER 5: THE THREE MCP PRIMITIVES IN DEPTH

Before we dive into building the server, let us spend serious time understanding what each MCP primitive is, how it works under the hood, and why you would choose one over another for a given piece of functionality.

Tools are the workhorses of MCP. A tool is a function that the model can decide to call. When the model determines that it needs to perform an action or retrieve parameterized information, it emits a special structured block in its response that says, in effect, "I want to call this tool with these arguments." The MCP client intercepts this, sends the request to the appropriate server, gets the result, appends it to the conversation as a tool result, and then asks the model to continue. The model sees the result and uses it to generate its next response. This is the fundamental agentic loop.

Tools are defined in FastMCP by decorating a Python function with @mcp.tool(). FastMCP inspects the function's type annotations and docstring to automatically generate a JSON schema that describes the tool's parameters. This schema is sent to the model as part of the tool list, and the model uses it to understand what arguments to provide when calling the tool.

Resources are a fundamentally different concept. A resource is not something the model calls on demand. It is a piece of data identified by a URI that the host application can fetch and inject into the model's context. Think of resources as the ambient information layer: data that is always potentially relevant and that the host proactively provides. In a system monitoring context, you might fetch the current CPU and memory state as resources at the start of every conversation, so the model already knows the system's baseline health before the user even asks a question.

Resources are defined with @mcp.resource("uri://scheme/path"). The URI is the identifier that clients use to request the resource. The function must return a string, which is the content of the resource. That content can be plain text, JSON, Markdown, or any other text format. FastMCP handles the rest. Resource URIs can also contain template variables in curly braces, like "system://disk/{path}", which FastMCP extracts and passes as function arguments.

Prompts are the third primitive, and they are the most conceptual. A prompt is a reusable template for a conversation or instruction. You define a prompt by decorating a function with @mcp.prompt(). The function receives parameters and returns a string that becomes a system message or user message in a conversation. Prompts are useful for standardizing how the model is instructed to approach certain tasks. For a system monitoring server, you might define a prompt called system_health_analyst that tells the model to examine the system metrics with a specific focus on anomalies and to respond in a structured format.

The relationship between these three primitives and the three kinds of control authority is worth memorizing, because it shapes every design decision you will make when building an MCP server. Tools are model-controlled, meaning the model decides when to call them based on its own reasoning. Resources are application-controlled, meaning the host application decides when to fetch them and inject them into context. Prompts are user-controlled, meaning the user explicitly selects a prompt to guide the conversation.

CHAPTER 6: BUILDING THE DOMAIN LAYER

Good software separates concerns. The MCP server should not contain raw psutil calls scattered throughout its tool and resource handlers. Instead, we create a dedicated domain layer that wraps psutil and provides a clean, well-typed interface. The MCP server then calls this domain layer. This makes the code testable, readable, and easy to modify if psutil's API ever changes.

The system_info.py module is that domain layer. It defines data classes for the various system metrics and provides functions that query psutil and return those data classes. Every function has a clear docstring, proper type annotations, and handles the edge cases that psutil can throw at you, particularly on different operating systems.

One important psutil behavior to understand before reading the code: when you call psutil.process_iter() with attrs=["cpu_percent"], the cpu_percent value for every process will be 0.0 on the first call. This is because cpu_percent() requires two measurements separated by a time interval to calculate a meaningful percentage. The first call establishes the baseline; only subsequent calls return real values. This is a fundamental limitation of how CPU measurement works at the operating system level. The list_processes() method documents this clearly and returns the data as-is, since the values are still useful for relative comparisons within a single snapshot and become accurate on repeated calls.

# server/system_info.py

#

# Domain layer: wraps the psutil library and provides clean, typed data

# structures for the MCP server to consume.

#

# Design principle: this module knows nothing about MCP. It is a pure Python

# wrapper around psutil. This separation makes it independently testable and

# keeps the MCP server layer focused on protocol concerns.

from __future__ import annotations

import json

import os

from dataclasses import asdict, dataclass

from typing import Optional

import psutil

@dataclass

class CpuInfo:

    """Snapshot of CPU utilization and frequency."""

    utilization_percent: float

    logical_core_count: int

    physical_core_count: int

    frequency_mhz: Optional[float]

    load_avg_1m: float

    load_avg_5m: float

    load_avg_15m: float

    def to_json(self) -> str:

        return json.dumps(asdict(self), indent=2)

@dataclass

class MemoryInfo:

    """Snapshot of virtual and swap memory usage."""

    total_bytes: int

    available_bytes: int

    used_bytes: int

    utilization_percent: float

    swap_total_bytes: int

    swap_used_bytes: int

    swap_utilization_percent: float

    def to_json(self) -> str:

        return json.dumps(asdict(self), indent=2)

@dataclass

class DiskInfo:

    """Disk usage for a specific mount point."""

    path: str

    total_bytes: int

    used_bytes: int

    free_bytes: int

    utilization_percent: float

    filesystem_type: str

    def to_json(self) -> str:

        return json.dumps(asdict(self), indent=2)

@dataclass

class ProcessSummary:

    """A lightweight summary of a running process."""

    pid: int

    name: str

    status: str

    cpu_percent: float

    memory_percent: float

    username: str

    def to_dict(self) -> dict:

        return asdict(self)

class SystemInfoService:

    """

    Provides read-only system information by querying psutil.

    All methods are pure queries: they observe system state but do not

    modify it. This class is the information-retrieval half of the domain.

    """

    def get_cpu_info(self) -> CpuInfo:

        """

        Collect a comprehensive CPU snapshot.

        The cpu_percent call uses a 0.1-second interval so it returns a

        meaningful measurement rather than 0.0 (which is what you get with

        interval=None on the very first call, because the function needs two

        measurements to calculate a percentage).

        """

        freq = psutil.cpu_freq()

        load_1, load_5, load_15 = psutil.getloadavg()

        return CpuInfo(

            utilization_percent=psutil.cpu_percent(interval=0.1),

            logical_core_count=psutil.cpu_count(logical=True),

            physical_core_count=psutil.cpu_count(logical=False) or 1,

            frequency_mhz=freq.current if freq else None,

            load_avg_1m=load_1,

            load_avg_5m=load_5,

            load_avg_15m=load_15,

        )

    def get_memory_info(self) -> MemoryInfo:

        """Collect virtual and swap memory statistics."""

        vm = psutil.virtual_memory()

        swap = psutil.swap_memory()

        return MemoryInfo(

            total_bytes=vm.total,

            available_bytes=vm.available,

            used_bytes=vm.used,

            utilization_percent=vm.percent,

            swap_total_bytes=swap.total,

            swap_used_bytes=swap.used,

            swap_utilization_percent=swap.percent,

        )

    def get_disk_info(self, path: str = "/") -> DiskInfo:

        """

        Collect disk usage statistics for the partition containing 'path'.

        Raises ValueError if the path does not exist on the filesystem.

        Uses longest-match logic to identify the correct partition when

        multiple partitions share a common path prefix (e.g., / and /var).

        """

        if not os.path.exists(path):

            raise ValueError(f"Path does not exist: {path}")

        usage = psutil.disk_usage(path)

        # Identify the filesystem type using longest-match logic.

        # A simple startswith() check would incorrectly match "/" for a

        # path like "/var" when both "/" and "/var" are mounted partitions.

        # By sorting on mountpoint length descending and taking the first

        # match, we always find the most specific (deepest) partition.

        fs_type = "unknown"

        best_match_len = 0

        for part in psutil.disk_partitions(all=False):

            mp = part.mountpoint

            if path.startswith(mp) and len(mp) > best_match_len:

                fs_type = part.fstype

                best_match_len = len(mp)

        return DiskInfo(

            path=path,

            total_bytes=usage.total,

            used_bytes=usage.used,

            free_bytes=usage.free,

            utilization_percent=usage.percent,

            filesystem_type=fs_type,

        )

    def list_processes(

        self,

        sort_by: str = "cpu_percent",

        limit: int = 20,

    ) -> list[ProcessSummary]:

        """

        Return a sorted list of running processes.

        The sort_by parameter accepts 'cpu_percent' or 'memory_percent'.

        The limit parameter caps the result to avoid overwhelming the model

        with hundreds of process entries.

        Important note on cpu_percent accuracy: psutil requires two

        measurements separated by a time interval to calculate a meaningful

        CPU percentage. When called via process_iter() with attrs=, the

        first call will return 0.0 for all processes. Values become accurate

        on subsequent calls. This is a fundamental OS-level measurement

        constraint, not a bug.

        """

        valid_sort_keys = {"cpu_percent", "memory_percent"}

        if sort_by not in valid_sort_keys:

            raise ValueError(

                f"sort_by must be one of {valid_sort_keys}, got '{sort_by}'"

            )

        processes: list[ProcessSummary] = []

        attrs = ["pid", "name", "status", "cpu_percent", "memory_percent", "username"]

        for proc in psutil.process_iter(attrs=attrs):

            try:

                info = proc.info

                processes.append(

                    ProcessSummary(

                        pid=info["pid"],

                        name=info["name"] or "unknown",

                        status=info["status"] or "unknown",

                        cpu_percent=info["cpu_percent"] or 0.0,

                        memory_percent=info["memory_percent"] or 0.0,

                        username=info["username"] or "unknown",

                    )

                )

            except (psutil.NoSuchProcess, psutil.AccessDenied):

                # Processes can disappear or deny access between the iter

                # call and the info access. This is normal; skip them.

                continue

        processes.sort(key=lambda p: getattr(p, sort_by), reverse=True)

        return processes[:limit]

class ProcessControlService:

    """

    Provides process management actions by calling psutil's Process API.

    All methods in this class have side effects: they modify the state of

    running processes. This is the action half of the domain layer. These

    operations map to MCP Tools rather than Resources.

    Security note: in a production deployment, these operations should be

    protected by authentication and authorization middleware. The MCP server

    layer is responsible for enforcing those controls.

    """

    def terminate_process(self, pid: int) -> str:

        """

        Send SIGTERM to the process with the given PID.

        SIGTERM is a graceful termination signal that allows the process

        to clean up before exiting. Most well-behaved processes respond to it.

        Returns a status message describing the outcome.

        """

        try:

            proc = psutil.Process(pid)

            proc_name = proc.name()

            proc.terminate()

            return (

                f"SIGTERM sent to process {pid} ({proc_name}). "

                f"It may take a moment to exit."

            )

        except psutil.NoSuchProcess:

            return f"No process found with PID {pid}."

        except psutil.AccessDenied:

            return f"Access denied: insufficient privileges to terminate PID {pid}."

    def kill_process(self, pid: int) -> str:

        """

        Send SIGKILL to the process with the given PID.

        SIGKILL is an immediate, unconditional termination signal. The process

        has no opportunity to clean up. Use this only when terminate() has

        failed or when immediate termination is required.

        """

        try:

            proc = psutil.Process(pid)

            proc_name = proc.name()

            proc.kill()

            return (

                f"SIGKILL sent to process {pid} ({proc_name}). "

                f"Process terminated immediately."

            )

        except psutil.NoSuchProcess:

            return f"No process found with PID {pid}."

        except psutil.AccessDenied:

            return f"Access denied: insufficient privileges to kill PID {pid}."

    def suspend_process(self, pid: int) -> str:

        """

        Suspend (pause) the process with the given PID by sending SIGSTOP.

        A suspended process is frozen in place: it consumes no CPU but

        remains in memory and retains all its state. It can be resumed later.

        """

        try:

            proc = psutil.Process(pid)

            proc_name = proc.name()

            proc.suspend()

            return (

                f"Process {pid} ({proc_name}) has been suspended "

                f"(SIGSTOP sent)."

            )

        except psutil.NoSuchProcess:

            return f"No process found with PID {pid}."

        except psutil.AccessDenied:

            return f"Access denied: insufficient privileges to suspend PID {pid}."

    def resume_process(self, pid: int) -> str:

        """

        Resume a previously suspended process by sending SIGCONT.

        The process will continue executing from exactly where it was frozen.

        """

        try:

            proc = psutil.Process(pid)

            proc_name = proc.name()

            proc.resume()

            return (

                f"Process {pid} ({proc_name}) has been resumed "

                f"(SIGCONT sent)."

            )

        except psutil.NoSuchProcess:

            return f"No process found with PID {pid}."

        except psutil.AccessDenied:

            return f"Access denied: insufficient privileges to resume PID {pid}."

    def set_process_priority(self, pid: int, niceness: int) -> str:

        """

        Change the scheduling priority (niceness) of a process.

        Niceness ranges from -20 (highest priority) to 19 (lowest priority).

        The default niceness for most processes is 0. Increasing niceness

        makes a process more polite to other processes; it yields CPU time

        more readily. Decreasing niceness (negative values) makes a process

        more aggressive about claiming CPU time, but typically requires root.

        """

        if not (-20 <= niceness <= 19):

            return (

                f"Invalid niceness value {niceness}. "

                f"Must be between -20 and 19 inclusive."

            )

        try:

            proc = psutil.Process(pid)

            proc_name = proc.name()

            proc.nice(niceness)

            return (

                f"Process {pid} ({proc_name}) niceness set to {niceness}."

            )

        except psutil.NoSuchProcess:

            return f"No process found with PID {pid}."

        except psutil.AccessDenied:

            return (

                f"Access denied: setting negative niceness values "

                f"requires root privileges."

            )

The domain layer is now complete and entirely independent of MCP. You could write unit tests for SystemInfoService and ProcessControlService without any MCP infrastructure at all. This is exactly the kind of separation that makes software maintainable over time.

CHAPTER 7: BUILDING THE MCP SERVER

With the domain layer in place, the MCP server becomes a thin translation layer. Its only job is to map MCP protocol requests to domain layer calls and format the results appropriately. The server.py file defines the FastMCP application, registers all the tools, resources, and prompts, and then exposes the whole thing as a FastAPI application.

Let us walk through the server construction piece by piece, because each section illustrates an important concept.

The server begins with imports and initialization. The FastMCP constructor takes the server's name, which is reported to clients during capability discovery, and the stateless_http=True flag, which tells the SDK to operate in the stateless mode required by the 2026-07-28 specification. We also instantiate our two domain service objects here, at module level, so they are shared across all requests. We also define a /health route on the FastAPI application before mounting the MCP server, so health checks remain accessible even after the MCP app is mounted.

# server/server.py

#

# The MCP server for the System Monitor application.

#

# This module is responsible for:

#   1. Defining MCP tools (actions with side effects)

#   2. Defining MCP resources (read-only data endpoints)

#   3. Defining MCP prompts (reusable instruction templates)

#   4. Exposing the whole thing as a stateless HTTP service at /mcp

#

# It delegates all actual work to the domain layer in system_info.py.

# This module should contain zero business logic.

from __future__ import annotations

import json

from contextlib import asynccontextmanager

from fastapi import FastAPI

from fastapi.responses import JSONResponse

from mcp.server.fastmcp import FastMCP

from .system_info import ProcessControlService, SystemInfoService

# ---------------------------------------------------------------------------

# Server initialization

# ---------------------------------------------------------------------------

# stateless_http=True implements the 2026-07-28 MCP specification.

# json_response=True tells FastMCP to serialize all responses as JSON,

# which is the most interoperable format for clients.

mcp = FastMCP(

    "System Monitor MCP Server",

    stateless_http=True,

    json_response=True,

)

# Shared domain service instances. These are lightweight objects that

# make no connections and hold no state, so sharing them is safe.

_system_info = SystemInfoService()

_process_control = ProcessControlService()

# ---------------------------------------------------------------------------

# MCP Resources: read-only data endpoints

#

# Resources are fetched by the host application to provide ambient context

# to the model. They do not require model-side invocation; the host decides

# when to fetch them.

# ---------------------------------------------------------------------------

@mcp.resource("system://cpu")

def resource_cpu_info() -> str:

    """

    Current CPU utilization, core count, frequency, and load averages.

    This resource is suitable for injecting as ambient context at the start

    of a system monitoring conversation. The model can reference this data

    without having to explicitly call a tool.

    """

    cpu = _system_info.get_cpu_info()

    return cpu.to_json()

@mcp.resource("system://memory")

def resource_memory_info() -> str:

    """

    Current virtual and swap memory usage statistics.

    Provides total, used, available, and utilization percentage for both

    physical RAM and swap space.

    """

    memory = _system_info.get_memory_info()

    return memory.to_json()

@mcp.resource("system://disk/{path}")

def resource_disk_info(path: str = "") -> str:

    """

    Disk usage for the partition containing the specified path.

    The path parameter is extracted from the URI template. For example,

    fetching system://disk/var returns disk statistics for the partition

    that contains /var. An empty path segment defaults to the root partition.

    """

    # The URI template strips the leading slash from the path segment.

    # We restore it here so psutil receives a valid absolute path.

    # An empty string means the client requested system://disk/ which

    # should map to the root partition.

    normalized_path = f"/{path}" if path and not path.startswith("/") else "/"

    try:

        disk = _system_info.get_disk_info(normalized_path)

        return disk.to_json()

    except ValueError as e:

        return json.dumps({"error": str(e)})

# ---------------------------------------------------------------------------

# MCP Tools: executable actions

#

# Tools are invoked by the model when it determines that an action is needed.

# Each tool's docstring is sent to the model as part of the tool description,

# so clear, precise documentation is essential for correct model behavior.

#

# The function's type annotations define the JSON schema for the tool's

# parameters. FastMCP generates this schema automatically.

# ---------------------------------------------------------------------------

@mcp.tool()

def list_processes(

    sort_by: str = "cpu_percent",

    limit: int = 20,

) -> str:

    """

    List the top running processes sorted by resource usage.

    Use this tool when the user asks about what is consuming CPU or memory,

    which processes are running, or what the most resource-intensive programs

    are. Returns a JSON array of process summaries.

    Note: cpu_percent values may show 0.0 on the very first call to this

    tool in a session. This is a known OS-level measurement constraint;

    values become accurate on subsequent calls.

    Args:

        sort_by: The metric to sort by. Must be 'cpu_percent' or

                 'memory_percent'. Defaults to 'cpu_percent'.

        limit:   Maximum number of processes to return. Defaults to 20.

                 Keep this number reasonable to avoid overwhelming the context.

    """

    try:

        processes = _system_info.list_processes(sort_by=sort_by, limit=limit)

        result = [p.to_dict() for p in processes]

        return json.dumps(result, indent=2)

    except ValueError as e:

        return json.dumps({"error": str(e)})

@mcp.tool()

def terminate_process(pid: int) -> str:

    """

    Send a graceful termination signal (SIGTERM) to a process.

    Use this tool when the user wants to stop a process gracefully, giving

    it a chance to save state and clean up resources. This is the preferred

    way to stop a process. If the process does not respond, use kill_process

    instead.

    Args:

        pid: The numeric Process ID (PID) of the process to terminate.

             You can find PIDs using the list_processes tool.

    """

    return _process_control.terminate_process(pid)

@mcp.tool()

def kill_process(pid: int) -> str:

    """

    Forcefully kill a process immediately (SIGKILL).

    Use this tool only when terminate_process has failed or when the user

    explicitly requests an immediate, unconditional kill. The process will

    have no opportunity to clean up. Data loss may occur.

    Args:

        pid: The numeric Process ID (PID) of the process to kill.

    """

    return _process_control.kill_process(pid)

@mcp.tool()

def suspend_process(pid: int) -> str:

    """

    Pause (freeze) a process without terminating it.

    The process will stop executing and consume no CPU, but it will remain

    in memory with all its state intact. Use this to temporarily free up

    CPU without losing the process. Resume it later with resume_process.

    Args:

        pid: The numeric Process ID (PID) of the process to suspend.

    """

    return _process_control.suspend_process(pid)

@mcp.tool()

def resume_process(pid: int) -> str:

    """

    Resume a previously suspended process.

    The process will continue executing from exactly where it was paused.

    Use this after suspend_process to bring a process back to life.

    Args:

        pid: The numeric Process ID (PID) of the process to resume.

    """

    return _process_control.resume_process(pid)

@mcp.tool()

def set_process_priority(pid: int, niceness: int) -> str:

    """

    Change the CPU scheduling priority of a process.

    Niceness ranges from -20 (highest priority, most aggressive) to 19

    (lowest priority, most polite). The default is 0. Use positive values

    to make a background process more respectful of other processes. Setting

    negative values requires root/administrator privileges.

    Args:

        pid:      The numeric Process ID (PID) of the target process.

        niceness: The new niceness value, between -20 and 19 inclusive.

    """

    return _process_control.set_process_priority(pid, niceness)

# ---------------------------------------------------------------------------

# MCP Prompts: reusable instruction templates

#

# Prompts are selected by the user or host application to configure the

# model's behavior. They are not called by the model; they are injected

# into the conversation as structured instructions.

# ---------------------------------------------------------------------------

@mcp.prompt()

def system_health_analyst() -> str:

    """

    Configure the model as a senior system reliability engineer.

    This prompt instructs the model to approach system monitoring questions

    with a structured, professional methodology: gather metrics first,

    identify anomalies, explain root causes, and recommend actions.

    """

    return (

        "You are a senior site reliability engineer with 15 years of experience "

        "in Linux systems administration and performance analysis. "

        "When analyzing system health, always follow this methodology: "

        "1. First, gather current CPU, memory, and disk metrics using the "

        "available resources and tools. "

        "2. Identify any values that exceed normal thresholds (CPU > 80%, "

        "memory > 85%, disk > 90% are warning levels). "

        "3. Investigate the top resource consumers using list_processes. "

        "4. Explain your findings in clear, non-technical language, then "

        "provide specific, actionable recommendations. "

        "5. Before taking any action (terminating or modifying processes), "

        "always explain what you intend to do and why, and ask for confirmation."

    )

@mcp.prompt()

def process_investigator(process_name: str) -> str:

    """

    Configure the model to investigate a specific process.

    Args:

        process_name: The name of the process to investigate.

    """

    return (

        f"You are investigating the process named '{process_name}'. "

        f"Use list_processes to find all instances of '{process_name}', "

        f"examine their CPU and memory consumption, and determine whether "

        f"their resource usage is normal or anomalous. "

        f"Provide a detailed analysis of what '{process_name}' does, "

        f"why it might be consuming the resources it is consuming, "

        f"and what options are available if intervention is needed."

    )

# ---------------------------------------------------------------------------

# FastAPI integration: expose the MCP server over HTTP

#

# The 2026-07-28 stateless MCP specification is served via Streamable HTTP.

# The MCP app is mounted at /mcp, leaving the root path free for health

# checks and other FastAPI routes. The official MCP SDK Client connects

# to http://localhost:8000/mcp.

# ---------------------------------------------------------------------------

@asynccontextmanager

async def lifespan(app: FastAPI):

    """

    Manage the lifecycle of the MCP session manager.

    FastMCP's session manager handles connection pooling and cleanup for

    the underlying MCP transport. It must be started before any requests

    are served and stopped cleanly when the server shuts down.

    """

    async with mcp.session_manager.run():

        yield

# The outer FastAPI app handles routing, middleware, and health checks.

app = FastAPI(

    title="System Monitor MCP Server",

    description=(

        "An MCP server that exposes system monitoring and process management "

        "capabilities from the psutil library."

    ),

    version="1.0.0",

    lifespan=lifespan,

)

@app.get("/health")

async def health_check():

    """

    Simple liveness probe endpoint.

    This route is defined BEFORE the MCP app is mounted so that it

    remains accessible at the root FastAPI level. Load balancers and

    container orchestrators can poll this endpoint to verify the server

    is running.

    """

    return JSONResponse({"status": "ok", "server": "System Monitor MCP Server"})

# Mount the MCP protocol handler at /mcp.

# All MCP requests (tool calls, resource fetches, prompt retrievals) will

# be routed through this ASGI application. Clients connect to:

# http://localhost:8000/mcp

app.mount("/mcp", mcp.streamable_http_app())

The entry point for running the server is a separate file, which keeps the server module clean and importable without side effects.

# run_server.py

#

# Entry point for starting the System Monitor MCP Server.

# Run this file from the project root with: python run_server.py

#

# The server listens on http://0.0.0.0:8000

# MCP endpoint:    http://localhost:8000/mcp

# Health check:    http://localhost:8000/health

import uvicorn

from server.server import app

if __name__ == "__main__":

    uvicorn.run(

        app,

        host="0.0.0.0",

        port=8000,

        log_level="info",

    )

CHAPTER 8: BUILDING THE SHARED MCP CLIENT LAYER

Both the local and remote LLM clients need to communicate with the MCP server. Rather than duplicating this code in each client module, we extract it into a shared mcp_client.py module. This module uses the official MCP Python SDK v2 Client class, which handles all the protocol details automatically: JSON-RPC framing, Streamable HTTP transport, capability negotiation, and response parsing.

Understanding this layer is important because it demystifies what happens between the model and the MCP server. The official Client class connects to the server's /mcp endpoint using the Streamable HTTP transport defined in the 2026-07-28 specification. It operates as an async context manager, establishing a connection on entry and releasing it cleanly on exit. All operations are async and must be awaited.

The design of mcp_client.py provides a thin convenience wrapper around the official Client that normalizes the return types into plain Python strings and lists, making it easier for the LLM client modules to consume without dealing with SDK-specific response objects.

# client/mcp_client.py

#

# Async MCP client wrapper using the official MCP Python SDK v2.

#

# This module wraps mcp.Client to provide a clean, consistent interface

# for the LLM client modules. It handles:

#   - Connecting to the MCP server via Streamable HTTP

#   - Discovering available tools (list_tools)

#   - Fetching resources (read_resource)

#   - Invoking tools (call_tool)

#

# All methods are async. Callers must use 'await' and run inside an

# async context (asyncio.run() or an existing event loop).

from __future__ import annotations

from typing import Any

from mcp import Client

from mcp.client.streamable_http import streamablehttp_client

class McpClient:

    """

    Async wrapper around the official MCP SDK v2 Client.

    Uses the Streamable HTTP transport to connect to the MCP server.

    The server must be running and accessible at the provided URL.

    Usage pattern (inside an async function):

        client = McpClient("http://localhost:8000/mcp")

        tools = await client.list_tools()

        result = await client.call_tool("list_processes", {"limit": 10})

    """

    def __init__(self, server_url: str = "http://localhost:8000/mcp") -> None:

        """

        Initialize the client with the MCP server's endpoint URL.

        Args:

            server_url: Full URL to the MCP server's /mcp endpoint.

                        Must include the /mcp path; the server mounts

                        the MCP app at that path.

        """

        self._server_url = server_url

    async def list_tools(self) -> list[dict[str, Any]]:

        """

        Retrieve the list of all tools the server exposes.

        Returns a list of dicts, each containing 'name', 'description',

        and 'inputSchema' fields. The inputSchema is a JSON Schema object

        that describes the tool's parameters.

        """

        async with streamablehttp_client(self._server_url) as (read, write, _):

            async with Client(read, write) as client:

                response = await client.list_tools()

                return [

                    {

                        "name": tool.name,

                        "description": tool.description or "",

                        "inputSchema": tool.inputSchema or {

                            "type": "object",

                            "properties": {},

                        },

                    }

                    for tool in response.tools

                ]

    async def read_resource(self, uri: str) -> str:

        """

        Fetch the content of a resource by its URI.

        The server returns the resource content as a string. The content

        format depends on the resource (JSON, plain text, etc.).

        Args:

            uri: The resource URI, e.g., 'system://cpu' or 'system://memory'.

        """

        async with streamablehttp_client(self._server_url) as (read, write, _):

            async with Client(read, write) as client:

                response = await client.read_resource(uri)

                if response.contents:

                    content = response.contents[0]

                    # TextResourceContents has a .text attribute;

                    # BlobResourceContents has a .blob attribute.

                    return getattr(content, "text", "") or ""

                return ""

    async def call_tool(

        self,

        tool_name: str,

        arguments: dict[str, Any],

    ) -> str:

        """

        Invoke a tool on the MCP server and return its result as a string.

        Args:

            tool_name: The name of the tool to invoke, e.g., 'list_processes'.

            arguments: A dict of arguments matching the tool's inputSchema.

        Returns:

            The tool's result as a string. For structured data, this will

            typically be a JSON string that the caller can parse.

        """

        async with streamablehttp_client(self._server_url) as (read, write, _):

            async with Client(read, write) as client:

                response = await client.call_tool(tool_name, arguments)

                # Tool results are a list of content items. We concatenate

                # all text items into a single string for simplicity.

                parts: list[str] = []

                for item in response.content:

                    text = getattr(item, "text", None)

                    if text:

                        parts.append(text)

                return "\n".join(parts)

The McpClient opens a fresh Streamable HTTP connection for each operation. This is correct and efficient for the stateless 2026-07-28 protocol, where each request is self-contained. In a high-throughput production scenario you could optimize by reusing a single connection across multiple calls within one agent turn, but for a tutorial this per-call pattern is clear and correct.

CHAPTER 9: THE ANTHROPIC CLIENT - CLAUDE SONNET 5

The Anthropic client demonstrates how to use a remote LLM with MCP. The pattern is a classic agentic loop: send the user's message to the model along with the list of available tools, check whether the model wants to call a tool, execute that tool via the MCP server, append the result to the conversation, and repeat until the model produces a final response without any tool calls.

Claude Sonnet 5, released on June 30, 2026, is Anthropic's production-grade model optimized for agentic workflows and tool use. It has a one-million-token context window, which means it can hold extremely long conversations and large tool results without losing context. Its tool-calling capabilities are mature and reliable.

Because the McpClient is async, the AnthropicMcpAgent must also be async. The __init__ method remains synchronous since it performs no I/O, but all methods that touch the MCP server are async. The main() function uses asyncio.run() to drive the async event loop from a synchronous entry point.

# client/client_anthropic.py

#

# MCP client using Claude Sonnet 5 (claude-sonnet-5-20260630) as the LLM.

#

# This module implements the full agentic tool-use loop:

#   1. Send user message + tool list to Claude Sonnet 5

#   2. If Claude requests a tool call, execute it via the MCP server

#   3. Append the tool result to the conversation

#   4. Repeat until Claude produces a final text response

#

# The MCP server runs separately (python run_server.py) and is accessed

# at http://localhost:8000/mcp by the McpClient.

from __future__ import annotations

import asyncio

import os

from typing import Any

import anthropic

from dotenv import load_dotenv

from .mcp_client import McpClient

# Load the ANTHROPIC_API_KEY from the .env file if present.

load_dotenv()

class AnthropicMcpAgent:

    """

    An agentic LLM client that connects Claude Sonnet 5 to an MCP server.

    This class manages the full conversation lifecycle, including the

    multi-turn tool-use loop that is the heart of agentic AI behavior.

    """

    # The model identifier for Claude Sonnet 5.

    # This is the balanced production model as of September 2026,

    # optimized for cost-effective agentic workflows and tool use.

    MODEL = "claude-sonnet-5-20260630"

    def __init__(

        self,

        mcp_server_url: str = "http://localhost:8000/mcp",

        system_prompt: str | None = None,

    ) -> None:

        """

        Initialize the agent.

        Args:

            mcp_server_url: Full URL to the running MCP server's /mcp endpoint.

            system_prompt:  Optional system prompt to configure the model's

                            behavior. If None, a sensible default is used.

        """

        self._anthropic = anthropic.Anthropic(

            api_key=os.environ["ANTHROPIC_API_KEY"]

        )

        self._mcp = McpClient(server_url=mcp_server_url)

        self._system_prompt = system_prompt or (

            "You are a helpful system monitoring assistant. "

            "You have access to tools that can inspect and manage system "

            "processes. Use them to answer the user's questions accurately. "

            "Always explain what you are doing and why before taking any "

            "action that modifies system state."

        )

    async def _fetch_tools(self) -> list[dict[str, Any]]:

        """

        Retrieve the tool list from the MCP server and format it for

        the Anthropic API.

        The Anthropic API expects tools with 'name', 'description', and

        'input_schema' fields. FastMCP generates schemas that are compatible

        with this format; we just rename 'inputSchema' to 'input_schema'.

        Raises RuntimeError with a clear message if the MCP server is

        not reachable, so callers receive actionable feedback.

        """

        try:

            raw_tools = await self._mcp.list_tools()

        except Exception as exc:

            raise RuntimeError(

                f"Could not connect to MCP server at {self._mcp._server_url}. "

                f"Is the server running? Start it with: python run_server.py\n"

                f"Underlying error: {exc}"

            ) from exc

        # The Anthropic SDK uses snake_case 'input_schema'; FastMCP returns

        # camelCase 'inputSchema'. We normalize here.

        return [

            {

                "name": tool["name"],

                "description": tool.get("description", ""),

                "input_schema": tool.get("inputSchema", {"type": "object"}),

            }

            for tool in raw_tools

        ]

    async def _get_context_resources(self) -> str:

        """

        Fetch ambient system context from MCP resources to inject into the

        conversation. This gives Claude baseline system state before the

        user even asks a question.

        Returns an empty string silently if resource fetching fails, so

        the conversation can proceed without ambient context rather than

        failing entirely.

        """

        try:

            cpu_data = await self._mcp.read_resource("system://cpu")

            memory_data = await self._mcp.read_resource("system://memory")

            return (

                f"Current system state:\n"

                f"CPU: {cpu_data}\n"

                f"Memory: {memory_data}"

            )

        except Exception:

            return ""

    async def chat(

        self,

        user_message: str,

        inject_context: bool = True,

    ) -> str:

        """

        Send a message and run the full agentic tool-use loop.

        Args:

            user_message:    The user's natural language request.

            inject_context:  If True, fetch system resources and prepend

                             them to the system prompt as ambient context.

        Returns:

            The model's final text response after all tool calls are resolved.

        """

        # Fetch the tool list fresh for each conversation turn so that

        # any tools added to the server are immediately available.

        tools = await self._fetch_tools()

        # Build the effective system prompt, optionally augmented with

        # current system state fetched from MCP resources.

        effective_system = self._system_prompt

        if inject_context:

            context = await self._get_context_resources()

            if context:

                effective_system = f"{self._system_prompt}\n\n{context}"

        messages: list[dict[str, Any]] = [

            {"role": "user", "content": user_message}

        ]

        # The agentic loop: keep calling the model until it stops requesting

        # tool calls and produces a final text response.

        while True:

            response = self._anthropic.messages.create(

                model=self.MODEL,

                max_tokens=4096,

                system=effective_system,

                tools=tools,

                messages=messages,

            )

            # Append the assistant's full response to the conversation history.

            messages.append({

                "role": "assistant",

                "content": response.content,

            })

            if response.stop_reason == "end_turn":

                # The model is done. Extract and return the final text block.

                for block in response.content:

                    if block.type == "text":

                        return block.text

                return ""

            elif response.stop_reason == "tool_use":

                # The model wants to call one or more tools.

                # Process every tool_use block in this response.

                tool_results = []

                for block in response.content:

                    if block.type != "tool_use":

                        continue

                    print(f"  [Tool call] {block.name}({block.input})")

                    # Execute the tool via the MCP server.

                    result = await self._mcp.call_tool(

                        tool_name=block.name,

                        arguments=block.input,

                    )

                    print(f"  [Tool result] {result[:200]}...")

                    tool_results.append({

                        "type": "tool_result",

                        "tool_use_id": block.id,

                        "content": result,

                    })

                # Tool results are returned as a 'user' role message.

                # This is the Anthropic API's convention: the model's tool

                # requests are in the 'assistant' turn; the results come

                # back in the next 'user' turn.

                messages.append({

                    "role": "user",

                    "content": tool_results,

                })

                # Loop back to call the model again with the tool results.

            else:

                return f"Unexpected stop reason: {response.stop_reason}"

async def _run_interactive() -> None:

    """Async implementation of the interactive chat loop."""

    print("System Monitor Agent (Claude Sonnet 5)")

    print("=" * 50)

    print("Connected to MCP server at http://localhost:8000/mcp")

    print("Type 'quit' to exit.\n")

    agent = AnthropicMcpAgent()

    while True:

        try:

            user_input = input("You: ").strip()

        except (EOFError, KeyboardInterrupt):

            print("\nGoodbye!")

            break

        if user_input.lower() in ("quit", "exit", "q"):

            print("Goodbye!")

            break

        if not user_input:

            continue

        try:

            response = await agent.chat(user_input)

            print(f"Agent: {response}\n")

        except RuntimeError as exc:

            print(f"Error: {exc}\n")

def main() -> None:

    """

    Interactive command-line interface for the Anthropic MCP agent.

    Run this function to start a conversation with Claude Sonnet 5

    backed by the System Monitor MCP server.

    """

    asyncio.run(_run_interactive())

if __name__ == "__main__":

    main()

The agentic loop in the chat() method is worth examining carefully, because it is the same pattern that underlies every sophisticated AI agent, regardless of which model or framework you use. The model receives the user's message and the list of available tools. It responds with either a final text answer or a request to call one or more tools. If it requests tool calls, you execute those calls, collect the results, and feed them back to the model as a new message. The model then either produces its final answer or requests more tool calls. This continues until the model is satisfied.

The inject_context parameter demonstrates how MCP resources integrate into this flow. Before the conversation starts, the client fetches the current CPU and memory state from the MCP server's resources and injects that data into the system prompt. This means Claude already knows the system's baseline health when it starts reasoning about the user's question. It does not need to call a tool just to find out the current memory usage; it already has that information.

CHAPTER 10: THE OLLAMA CLIENT - LOCAL LLMS

The Ollama client follows the same pattern as the Anthropic client, but uses a locally running model instead of a remote API. This is important for several reasons. Privacy-sensitive environments may not allow data to leave the local network. Cost-conscious deployments may prefer to run inference locally rather than paying per token. Development and testing environments benefit from faster iteration without API rate limits.

Ollama's Python SDK exposes a Client class with a chat() method that accepts a tools parameter. The SDK returns response objects with attribute-based access rather than dictionary access. Specifically, the response from chat() is a ChatResponse object, and the message within it is a Message object. You access these with response.message and response.message.tool_calls, not with dictionary subscript syntax. Tool calls themselves are objects with a .function attribute containing .name and .arguments properties.

The messages list that you build up over the conversation uses plain Python dicts with role and content keys, which is the format the Ollama SDK expects when you pass the history back on subsequent calls. When you receive a Message object from the response, you convert it to a dict before appending it to the messages list. Tool results are appended as dicts with role "tool" and the result string as the content value.

# client/client_ollama.py

#

# MCP client using a local Ollama model (llama3.3 by default).

#

# This module implements the same agentic tool-use loop as the Anthropic

# client, but uses a locally running model via the Ollama Python SDK.

#

# Prerequisites:

#   - Ollama installed and running (ollama serve)

#   - The target model pulled (ollama pull llama3.3)

#   - MCP server running (python run_server.py)

#

# The Ollama SDK uses the OpenAI tool-calling format, which differs slightly

# from the Anthropic format. Tool results use role "tool" rather than being

# embedded in a "user" role message.

from __future__ import annotations

import asyncio

import json

from typing import Any

import ollama

from dotenv import load_dotenv

from .mcp_client import McpClient

load_dotenv()

class OllamaMcpAgent:

    """

    An agentic LLM client that connects a local Ollama model to an MCP server.

    Uses the same tool-use loop pattern as AnthropicMcpAgent, adapted

    for the Ollama Python SDK's API conventions. All MCP operations are

    async; Ollama SDK calls are synchronous (the SDK does not yet expose

    a native async interface, so we call it from within the async loop).

    """

    def __init__(

        self,

        model: str = "llama3.3",

        mcp_server_url: str = "http://localhost:8000/mcp",

        ollama_host: str = "http://localhost:11434",

    ) -> None:

        """

        Initialize the agent.

        Args:

            model:          The Ollama model name to use. Must be pulled

                            locally before use (ollama pull <model>).

            mcp_server_url: Full URL to the running MCP server's /mcp endpoint.

            ollama_host:    URL of the Ollama server. Defaults to localhost.

                            Change this if Ollama runs on a different machine

                            or in a container.

        """

        self._model = model

        self._mcp = McpClient(server_url=mcp_server_url)

        # The Ollama Client accepts a host parameter to point at a non-default

        # Ollama instance, enabling remote or containerized deployments.

        self._ollama = ollama.Client(host=ollama_host)

    async def _fetch_tools_for_ollama(self) -> list[dict[str, Any]]:

        """

        Retrieve tools from the MCP server and convert them to the

        OpenAI/Ollama tool-calling format.

        Ollama expects tools in this structure:

            {

                "type": "function",

                "function": {

                    "name": "...",

                    "description": "...",

                    "parameters": { ... JSON Schema ... }

                }

            }

        FastMCP returns tools with 'inputSchema' at the top level.

        We wrap them in the 'function' envelope that Ollama expects.

        Raises RuntimeError with a clear message if the MCP server is

        not reachable.

        """

        try:

            raw_tools = await self._mcp.list_tools()

        except Exception as exc:

            raise RuntimeError(

                f"Could not connect to MCP server at {self._mcp._server_url}. "

                f"Is the server running? Start it with: python run_server.py\n"

                f"Underlying error: {exc}"

            ) from exc

        return [

            {

                "type": "function",

                "function": {

                    "name": tool["name"],

                    "description": tool.get("description", ""),

                    "parameters": tool.get(

                        "inputSchema",

                        {"type": "object", "properties": {}},

                    ),

                },

            }

            for tool in raw_tools

        ]

    async def _get_context_resources(self) -> str:

        """Fetch ambient system state from MCP resources."""

        try:

            cpu_data = await self._mcp.read_resource("system://cpu")

            memory_data = await self._mcp.read_resource("system://memory")

            return (

                f"Current system state at conversation start:\n"

                f"CPU metrics: {cpu_data}\n"

                f"Memory metrics: {memory_data}"

            )

        except Exception:

            return ""

    @staticmethod

    def _message_to_dict(msg: Any) -> dict[str, Any]:

        """

        Convert an Ollama Message object to a plain dict for the messages list.

        The Ollama SDK returns Message objects from chat(), but expects

        plain dicts when you pass the conversation history back. This helper

        performs the conversion, preserving tool_calls if present.

        """

        d: dict[str, Any] = {

            "role": msg.role,

            "content": msg.content or "",

        }

        if msg.tool_calls:

            d["tool_calls"] = msg.tool_calls

        return d

    async def chat(

        self,

        user_message: str,

        inject_context: bool = True,

    ) -> str:

        """

        Send a message and run the full agentic tool-use loop with Ollama.

        Args:

            user_message:   The user's natural language request.

            inject_context: If True, inject ambient resource data as context.

        Returns:

            The model's final text response.

        """

        tools = await self._fetch_tools_for_ollama()

        # Build the system message, optionally augmented with resource context.

        system_content = (

            "You are a helpful system monitoring assistant. "

            "Use the available tools to answer questions about system "

            "performance and manage processes when asked. "

            "Always explain your reasoning before taking actions."

        )

        if inject_context:

            context = await self._get_context_resources()

            if context:

                system_content = f"{system_content}\n\n{context}"

        # Initialize the messages list with the system message and the

        # user's first message. Subsequent turns append to this list.

        messages: list[dict[str, Any]] = [

            {"role": "system", "content": system_content},

            {"role": "user", "content": user_message},

        ]

        # The agentic loop for Ollama.

        while True:

            # The Ollama SDK's chat() is synchronous. We call it directly

            # from within the async function; it blocks the event loop

            # briefly but is acceptable for a single-user CLI tool.

            response = self._ollama.chat(

                model=self._model,

                messages=messages,

                tools=tools,

            )

            # response.message is an ollama Message object (not a dict).

            # We convert it to a dict before appending to messages.

            assistant_msg = response.message

            messages.append(self._message_to_dict(assistant_msg))

            # Ollama signals tool calls through the .tool_calls attribute

            # on the Message object. If this attribute is None or an empty

            # list, the model has produced a final text response.

            tool_calls = assistant_msg.tool_calls

            if not tool_calls:

                # No tool calls: the model is done. Return the text content.

                return assistant_msg.content or ""

            # Process each tool call in the response.

            for tool_call in tool_calls:

                func = tool_call.function

                tool_name = func.name

                # In current Ollama SDK versions, arguments is a dict.

                # We guard against older versions that returned a JSON string.

                raw_args = func.arguments

                if isinstance(raw_args, str):

                    try:

                        arguments = json.loads(raw_args)

                    except json.JSONDecodeError:

                        arguments = {}

                else:

                    arguments = raw_args if raw_args is not None else {}

                print(f"  [Tool call] {tool_name}({arguments})")

                # Execute the tool via the MCP server (async call).

                result = await self._mcp.call_tool(

                    tool_name=tool_name,

                    arguments=arguments,

                )

                print(f"  [Tool result] {result[:200]}...")

                # Append the tool result as a 'tool' role message.

                # Ollama follows the OpenAI convention where tool results

                # use the 'tool' role rather than being embedded in 'user'.

                messages.append({

                    "role": "tool",

                    "content": result,

                })

            # Loop back to call the model with the tool results appended.

async def _run_interactive() -> None:

    """Async implementation of the interactive chat loop."""

    print("System Monitor Agent (Ollama / llama3.3)")

    print("=" * 50)

    print("Connected to Ollama at http://localhost:11434")

    print("Connected to MCP server at http://localhost:8000/mcp")

    print("Type 'quit' to exit.\n")

    agent = OllamaMcpAgent(model="llama3.3")

    while True:

        try:

            user_input = input("You: ").strip()

        except (EOFError, KeyboardInterrupt):

            print("\nGoodbye!")

            break

        if user_input.lower() in ("quit", "exit", "q"):

            print("Goodbye!")

            break

        if not user_input:

            continue

        try:

            response = await agent.chat(user_input)

            print(f"Agent: {response}\n")

        except RuntimeError as exc:

            print(f"Error: {exc}\n")

def main() -> None:

    """

    Interactive command-line interface for the Ollama MCP agent.

    Run this to start a conversation with a local Ollama model backed

    by the System Monitor MCP server.

    """

    asyncio.run(_run_interactive())

if __name__ == "__main__":

    main()

The key difference between the Anthropic and Ollama clients lies in the tool-calling format and the message role conventions. Anthropic uses a block.type == "tool_use" check in the assistant's response and expects tool results back in a "user" role message containing a list of tool_result dicts. Ollama follows the OpenAI convention: tool calls appear in a .tool_calls attribute on the Message object, and results are appended as "tool" role messages. Both achieve the same thing; they just use different wire formats for the same concept.

This is one of the reasons MCP is so valuable. The MCP server itself is completely agnostic to which client is calling it. The same server handles requests from Claude Sonnet 5, from Llama 3.3 via Ollama, or from any other MCP-compatible client. The translation between model-specific tool-calling formats and the MCP protocol is the client's responsibility, not the server's.

CHAPTER 11: THE CONFIGURATION MODULE

A production-ready application centralizes its configuration rather than scattering magic strings and environment variable reads throughout the codebase. The config.py module provides a single source of truth for all configurable parameters.

# config.py

#

# Centralized configuration for the System Monitor MCP application.

#

# All configurable values live here. Environment variables take precedence

# over defaults, making the application easy to deploy in different

# environments without code changes.

#

# Usage:

#   from config import SERVER, ANTHROPIC, OLLAMA

#   agent = AnthropicMcpAgent(mcp_server_url=SERVER.mcp_url)

from __future__ import annotations

import os

from dataclasses import dataclass, field

from dotenv import load_dotenv

# Load .env file if present. This is a no-op in production environments

# where variables are set directly in the environment.

load_dotenv()

@dataclass(frozen=True)

class ServerConfig:

    """Configuration for the MCP server."""

    host: str = field(

        default_factory=lambda: os.getenv("MCP_SERVER_HOST", "0.0.0.0")

    )

    port: int = field(

        default_factory=lambda: int(os.getenv("MCP_SERVER_PORT", "8000"))

    )

    @property

    def mcp_url(self) -> str:

        """

        The full URL of the MCP endpoint, as seen by clients on localhost.

        Note: host is 0.0.0.0 for server binding (accept all interfaces),

        but clients connect via localhost. These are intentionally different.

        """

        return f"http://localhost:{self.port}/mcp"

    @property

    def health_url(self) -> str:

        """The URL of the health check endpoint."""

        return f"http://localhost:{self.port}/health"

@dataclass(frozen=True)

class AnthropicConfig:

    """Configuration for the Anthropic client."""

    api_key: str = field(

        default_factory=lambda: os.environ.get("ANTHROPIC_API_KEY", "")

    )

    # Claude Sonnet 5: the production-grade, cost-effective model

    # optimized for agentic tool use. Released June 30, 2026.

    model: str = "claude-sonnet-5-20260630"

    max_tokens: int = 4096

@dataclass(frozen=True)

class OllamaConfig:

    """Configuration for the Ollama local LLM client."""

    host: str = field(

        default_factory=lambda: os.getenv("OLLAMA_HOST", "http://localhost:11434")

    )

    # llama3.3 offers strong tool-calling capabilities and runs well

    # on consumer hardware with 16 GB+ of RAM.

    model: str = field(

        default_factory=lambda: os.getenv("OLLAMA_MODEL", "llama3.3")

    )

# Module-level singletons for easy import throughout the application.

SERVER = ServerConfig()

ANTHROPIC = AnthropicConfig()

OLLAMA = OllamaConfig()

CHAPTER 12: RUNNING THE COMPLETE SYSTEM

Now that all the pieces are in place, let us walk through exactly how to start everything up and what a real interaction looks like.

Start the MCP server in one terminal window from the project root:

python run_server.py

You should see Uvicorn start and report that it is listening on port 8000. The server is now ready to accept MCP requests. You can verify it is working by checking the health endpoint first:

curl http://localhost:8000/health

The server will respond with:

{"status": "ok", "server": "System Monitor MCP Server"}

In a second terminal, start the Anthropic client from the project root:

python -m client.client_anthropic

Or start the Ollama client:

python -m client.client_ollama

Note the -m flag and the dotted module path. Running client modules as packages this way ensures that relative imports within the client package resolve correctly. Running python client/client_anthropic.py directly would fail because Python would not recognize the package structure.

Here is an example of what a real conversation looks like. The lines beginning with [Tool call] and [Tool result] are printed by the client to show the tool-use loop in action:

  You: What is consuming the most CPU on this system right now?

  [Tool call] list_processes({'sort_by': 'cpu_percent', 'limit': 10})

  [Tool result] [{"pid": 4821, "name": "python3", "status": "running",

                   "cpu_percent": 45.2, "memory_percent": 3.1, ...

  Agent: Based on the current process list, the top CPU consumer on your

  system is python3 (PID 4821), which is using 45.2% of CPU. This is

  likely a computational workload or a script running in a tight loop.

  The second highest consumer is chrome (PID 3201) at 12.8%.

  If the python3 process is unexpected or has been running for an

  unusually long time, you might want to investigate it further. Would

  you like me to get more details about that process, or take any action?

  You: That python3 process is stuck in an infinite loop. Please terminate it.

  Agent: I will send a graceful termination signal (SIGTERM) to process

  4821. This gives the process a chance to clean up before exiting.

  [Tool call] terminate_process({'pid': 4821})

  [Tool result] SIGTERM sent to process 4821 (python3). It may take a

                moment to exit.

  Agent: Done. I have sent SIGTERM to the python3 process (PID 4821).

  It should exit shortly. If it does not respond to the graceful signal

  within a few seconds, we can use kill_process to force an immediate

  termination. Would you like me to check the process list again to

  confirm it has exited?

This interaction illustrates the full power of the MCP architecture. The model received ambient context about the system from the resources injected at conversation start. It then used the list_processes tool to get current data. It reasoned about that data, explained its findings, and when asked to act, it chose the appropriate tool, explained what it was doing, and executed the action. The entire flow is natural, transparent, and correct.

CHAPTER 13: ADVANCED PATTERNS AND PRODUCTION CONSIDERATIONS

The server and clients you have built are fully functional, but a production deployment requires additional thought in several areas.

The first area is authentication and authorization. The MCP server as built accepts any request from any client. In production, you need to add authentication middleware to FastAPI. The simplest approach is API key authentication: clients include an Authorization header with a bearer token, and FastAPI middleware validates it before routing the request. For process control operations specifically, you should consider requiring elevated privileges or a separate, more strongly authenticated endpoint. The /health route is intentionally excluded from authentication so that load balancers can poll it freely.

# Authentication middleware for server/server.py

# Add this BEFORE the app.mount("/mcp", ...) call.

# Import this at the top of server.py alongside the other imports.

import os

from starlette.middleware.base import BaseHTTPMiddleware

from fastapi import Request, HTTPException

VALID_API_KEYS = {os.getenv("MCP_API_KEY", "dev-key-change-in-production")}

class ApiKeyMiddleware(BaseHTTPMiddleware):

    """

    Validates the Authorization header on every incoming MCP request.

    Clients must include: Authorization: Bearer <api-key>

    The /health endpoint is excluded so load balancers can poll it freely.

    Requests without a valid key receive a 403 Forbidden response.

    """

    async def dispatch(self, request: Request, call_next):

        # Allow health check endpoint without authentication.

        if request.url.path == "/health":

            return await call_next(request)

        auth_header = request.headers.get("Authorization", "")

        if not auth_header.startswith("Bearer "):

            raise HTTPException(

                status_code=401,

                detail="Authorization header missing or malformed. "

                       "Expected: Authorization: Bearer <api-key>",

            )

        token = auth_header.removeprefix("Bearer ").strip()

        if token not in VALID_API_KEYS:

            raise HTTPException(

                status_code=403,

                detail="Invalid API key.",

            )

        return await call_next(request)

# Add this line after creating the FastAPI app and before app.mount():

# app.add_middleware(ApiKeyMiddleware)

The second area is error handling and observability. The domain layer already handles psutil-specific exceptions like NoSuchProcess and AccessDenied. But you should also add structured logging throughout the server so you can trace exactly what tool calls were made, by which client, at what time, and with what results. Python's standard logging module works well for this; you can configure it to emit JSON-formatted log records that are easy to ingest into log aggregation systems.

The third area is tool result size management. Large language models have context windows measured in tokens, and every tool result consumes tokens. The list_processes tool already has a limit parameter, but you should think carefully about all your tools and ensure that no tool can return an unbounded amount of data. If a tool might return large results, consider adding pagination support or a max_bytes parameter that truncates the output with a note indicating truncation.

The fourth area is testing. Because the domain layer is completely decoupled from MCP, you can write unit tests for SystemInfoService and ProcessControlService using standard pytest without any MCP infrastructure. For the server layer, FastMCP provides a test client that lets you invoke tools and resources in-process without starting an HTTP server. For the client layer, you can mock the McpClient to test the agentic loop logic without hitting a real server.

# Example: Unit tests for the domain layer using pytest

# Save as tests/test_system_info.py and run with: pytest tests/

import pytest

from unittest.mock import MagicMock, patch

from server.system_info import CpuInfo, SystemInfoService

class TestSystemInfoService:

    """Tests for the read-only system information service."""

    def test_get_cpu_info_returns_cpu_info(self):

        """

        Verify that get_cpu_info returns a properly structured CpuInfo

        object with all required fields populated.

        We mock all psutil calls to make tests deterministic, fast, and

        independent of the actual hardware the tests run on.

        """

        service = SystemInfoService()

        with (

            patch("psutil.cpu_percent", return_value=42.5),

            patch("psutil.cpu_count", side_effect=[8, 4]),

            patch("psutil.cpu_freq", return_value=MagicMock(current=3600.0)),

            patch("psutil.getloadavg", return_value=(1.5, 2.0, 2.5)),

        ):

            result = service.get_cpu_info()

        assert isinstance(result, CpuInfo)

        assert result.utilization_percent == 42.5

        assert result.logical_core_count == 8

        assert result.physical_core_count == 4

        assert result.frequency_mhz == 3600.0

        assert result.load_avg_1m == 1.5

    def test_get_disk_info_raises_for_nonexistent_path(self):

        """

        Verify that get_disk_info raises ValueError when the path does

        not exist, rather than propagating a raw psutil error.

        This tests the input validation layer in the domain service.

        """

        service = SystemInfoService()

        with pytest.raises(ValueError, match="Path does not exist"):

            service.get_disk_info("/this/path/does/not/exist/anywhere")

    def test_get_disk_info_uses_longest_match_for_filesystem_type(self):

        """

        Verify that get_disk_info correctly identifies the filesystem type

        using longest-match logic when multiple partitions share a prefix.

        Without longest-match logic, a path like /var/log would incorrectly

        match the '/' partition instead of the '/var' partition.

        """

        service = SystemInfoService()

        mock_partitions = [

            MagicMock(mountpoint="/", fstype="ext4"),

            MagicMock(mountpoint="/var", fstype="xfs"),

        ]

        mock_usage = MagicMock(

            total=100_000_000_000,

            used=50_000_000_000,

            free=50_000_000_000,

            percent=50.0,

        )

        with (

            patch("os.path.exists", return_value=True),

            patch("psutil.disk_usage", return_value=mock_usage),

            patch("psutil.disk_partitions", return_value=mock_partitions),

        ):

            result = service.get_disk_info("/var/log/syslog")

        assert result.filesystem_type == "xfs"

```

The fifth area is deployment. The MCP server is a standard FastAPI ASGI application. You can deploy it anywhere you would deploy any Python web service: as a Docker container, as a systemd service, on Kubernetes, or on any cloud platform that supports Python web applications. Because the 2026-07-28 specification is fully stateless, you can run multiple instances behind a load balancer without any session affinity requirements. This is a significant operational advantage over earlier MCP versions.

A minimal but production-appropriate Dockerfile for the server looks like this:

# Dockerfile

#

# Multi-stage build for the System Monitor MCP Server.

# Stage 1 installs dependencies; Stage 2 creates a lean runtime image.

FROM python:3.12-slim AS builder

WORKDIR /build

COPY requirements.txt .

RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# ---------------------------------------------------------------------------

FROM python:3.12-slim AS runtime

WORKDIR /app

# Copy installed packages from the builder stage.

COPY --from=builder /install /usr/local

# Copy application source.

COPY server/ ./server/

COPY config.py .

COPY run_server.py .

# The MCP server does not need root privileges.

RUN useradd --no-create-home --shell /bin/false appuser

USER appuser

EXPOSE 8000

# Uvicorn serves the FastAPI app. The MCP endpoint is at /mcp.

CMD ["python", "run_server.py"]

Build and run the container with:

docker build -t system-monitor-mcp .

docker run -p 8000:8000 -e MCP_API_KEY=your-key system-monitor-mcp

The sixth and final area worth discussing is the Extensions framework introduced in the 2026-07-28 specification. Extensions are optional capabilities that a server can advertise. The most relevant for our use case is the Tasks extension, which supports long-running operations. If you were to add a tool that runs a system benchmark or collects performance data over a five-minute window, the Tasks extension lets you model this as an asynchronous job: the client submits the task, gets a task ID back immediately, and then polls for completion. This is far more robust than a synchronous tool call that blocks for five minutes and risks timing out.

CHAPTER 14: WHAT YOU HAVE BUILT AND WHERE TO GO NEXT

Step back and look at what you have created. You have built a complete MCP server that wraps psutil, a Python library with no REST interface, no HTTP endpoints, and no JSON serialization of its own. You have demonstrated how to map the library's information-retrieval methods to MCP resources and its action methods to MCP tools. You have built a shared async client layer using the official MCP Python SDK v2 that speaks the Streamable HTTP transport of the 2026-07-28 specification, and you have built two separate LLM clients: one using Claude Sonnet 5 via the Anthropic API for production-grade remote inference, and one using Llama 3.3 via Ollama for private, cost-free local inference. Both clients implement the same agentic tool-use loop and connect to the same MCP server.

The architecture you have implemented is genuinely production-ready in its structure, even if it needs the security and observability additions discussed in Chapter 13 before going live. More importantly, the pattern you have learned is universal. The same approach works for any Python library, any SDK, any system that is not REST-based. Replace psutil with the boto3 AWS SDK, and you have an MCP server that gives your AI access to cloud infrastructure. Replace it with the Kubernetes Python client, and you have an AI that can inspect and manage your cluster. Replace it with a database driver, and you have an AI that can query and update your data. The pattern is always the same: wrap the SDK in a clean domain layer, map read operations to resources and write operations to tools, decorate with FastMCP, and mount on FastAPI.

MCP has become the lingua franca of AI tool integration for a very good reason. It gives you a clean, testable, model-agnostic boundary between your AI and your systems. The model does not know or care whether it is talking to psutil or PostgreSQL or Kubernetes. It sees tools, resources, and prompts. The server does not know or care whether it is serving Claude Sonnet 5 or Llama 3.3 or any other model. It sees MCP requests. This separation is what makes the ecosystem composable, and composability is what makes it powerful.

The next steps from here are to explore the MCP Extensions framework for long-running tasks, to add authentication middleware, to wire up structured logging with a tool like OpenTelemetry, and to write a comprehensive test suite. You might also explore the TypeScript MCP SDK, which is particularly well-suited for building MCP servers that integrate with Node.js-based tools and web APIs. The protocol is the same; only the implementation language changes.

Welcome to the agentic age. Your AI now has hands.