Introduction to MCP and Local File Access
The Model Context Protocol (MCP) has emerged as a standardized way for AI assistants and applications to interact with external tools, data sources, and services. While many developers are familiar with using pre-built MCP servers, building a custom MCP client gives you fine-grained control over how your application communicates with MCP servers, handles responses, and manages the lifecycle of tool invocations. In this tutorial, we'll build a custom MCP client specifically designed for local file access — one of the most common and useful applications of the protocol.
What Is an MCP Client?
An MCP client is the consumer side of the Model Context Protocol. It connects to one or more MCP servers, discovers the tools and resources they expose, and invokes those tools on behalf of a user or application. Think of it as a bridge: on one side sits your application (perhaps an AI agent, a CLI tool, or a web app), and on the other side sits an MCP server that actually performs the work — reading files, querying databases, calling APIs, and so on.
The protocol itself is built on JSON-RPC 2.0 and typically runs over standard input/output (stdio) for local servers or over HTTP/SSE for remote servers. For local file access, the stdio transport is the natural choice because it keeps everything on the same machine with minimal overhead.
Client vs. Server: Understanding the Distinction
It's easy to confuse the two sides of MCP, so let's clarify:
- MCP Server: Exposes capabilities. A file-access server, for example, might expose tools like
read_file,write_file,list_directory, andsearch_files. - MCP Client: Consumes those capabilities. It manages the connection, sends requests, parses responses, and integrates the results into your application logic.
In this tutorial, we focus on the client. We'll assume a file-access MCP server already exists (or we'll write a minimal one for testing), and we'll build a robust client around it.
Why Build a Custom MCP Client?
You might wonder why you'd build a custom client when reference implementations exist in TypeScript and Python. There are several compelling reasons:
- Tailored error handling: Different applications need different retry strategies, timeout behaviors, and fallback logic.
- Custom permission flows: You may want to prompt users before sensitive operations like writing or deleting files.
- Integration with existing systems: A custom client can plug directly into your logging, telemetry, and authentication infrastructure.
- Language flexibility: If your stack isn't TypeScript or Python, you need a client in your language of choice.
- Learning and debugging: Understanding the protocol deeply makes you a better MCP consumer and contributor.
Local file access is an ideal starting point because it's concrete, immediately useful, and exercises the core mechanics of MCP without the complexity of network authentication or remote state.
Prerequisites and Setup
For this tutorial, we'll use Python because the official MCP Python SDK provides a solid foundation and is widely accessible. You'll need:
- Python 3.10 or later
- The
mcpPython package (version 1.0 or later) - Basic familiarity with
asyncioand JSON-RPC concepts
Install the dependencies in a virtual environment:
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
pip install "mcp[cli]"
We'll also create a project structure:
mcp-file-client/
├── client.py # The custom MCP client
├── server.py # A minimal file-access MCP server for testing
└── sandbox/ # A safe directory for file operations
└── sample.txt
Create the sandbox directory and a sample file so we have something to work with:
mkdir -p sandbox
echo "Hello from the local filesystem!" > sandbox/sample.txt
A Minimal File-Access MCP Server
Before building the client, let's create a small MCP server that exposes file operations. This gives us a concrete target to connect to and keeps the tutorial self-contained. In a real project, you might use an existing server, but understanding the server side helps you write a better client.
# server.py
import os
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
SANDBOX_DIR = os.path.join(os.path.dirname(__file__), "sandbox")
app = Server("file-access-server")
def safe_path(filename: str) -> str:
"""Resolve a filename within the sandbox, preventing path traversal."""
base = os.path.abspath(SANDBOX_DIR)
target = os.path.abspath(os.path.join(base, filename))
if not target.startswith(base):
raise ValueError(f"Path '{filename}' escapes the sandbox")
return target
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="read_file",
description="Read the contents of a file within the sandbox.",
inputSchema={
"type": "object",
"properties": {
"filename": {
"type": "string",
"description": "Relative path to the file inside the sandbox."
}
},
"required": ["filename"]
}
),
Tool(
name="list_directory",
description="List files and directories in the sandbox.",
inputSchema={
"type": "object",
"properties": {
"subdir": {
"type": "string",
"description": "Optional subdirectory within the sandbox."
}
}
}
),
Tool(
name="write_file",
description="Write content to a file within the sandbox.",
inputSchema={
"type": "object",
"properties": {
"filename": {"type": "string"},
"content": {"type": "string"}
},
"required": ["filename", "content"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "read_file":
path = safe_path(arguments["filename"])
with open(path, "r", encoding="utf-8") as f:
return [TextContent(type="text", text=f.read())]
elif name == "list_directory":
subdir = arguments.get("subdir", "")
path = safe_path(subdir)
entries = os.listdir(path)
return [TextContent(type="text", text=json.dumps(entries, indent=2))]
elif name == "write_file":
path = safe_path(arguments["filename"])
with open(path, "w", encoding="utf-8") as f:
f.write(arguments["content"])
return [TextContent(type="text", text=f"Wrote {len(arguments['content'])} bytes to {arguments['filename']}")]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
import asyncio
asyncio.run(main())
This server exposes three tools — read_file, list_directory, and write_file — all constrained to a sandbox directory. The safe_path helper prevents path traversal attacks, which is a critical security measure we'll discuss later.
Building the Custom MCP Client
Now for the main event. Our custom client will:
- Spawn the server as a subprocess and communicate over stdio
- Initialize the MCP session with proper handshake
- Discover available tools
- Invoke tools and parse responses
- Handle errors gracefully with timeouts and retries
- Clean up resources on shutdown
Step 1: Defining the Client Class
We'll structure the client as a class that manages the subprocess lifecycle and exposes high-level methods for each operation. This makes it easy to integrate into larger applications.
# client.py
import asyncio
import json
import logging
from contextlib import AsyncExitStack
from typing import Any, Optional
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
logger = logging.getLogger(__name__)
class FileMCPClient:
"""A custom MCP client for local file access via a stdio server."""
def __init__(self, server_command: list[str], timeout: float = 30.0):
self.server_command = server_command
self.timeout = timeout
self.session: Optional[ClientSession] = None
self._exit_stack: Optional[AsyncExitStack] = None
self._tools_cache: Optional[list[dict]] = None
async def connect(self) -> None:
"""Start the server subprocess and initialize the MCP session."""
self._exit_stack = AsyncExitStack()
server_params = StdioServerParameters(
command=self.server_command[0],
args=self.server_command[1:],
env=None # Inherit the current environment
)
# Open the stdio transport
stdio_transport = await self._exit_stack.enter_async_context(
stdio_client(server_params)
)
read_stream, write_stream = stdio_transport
# Create and initialize the client session
self.session = await self._exit_stack.enter_async_context(
ClientSession(read_stream, write_stream)
)
await self.session.initialize()
logger.info("MCP session initialized successfully")
async def disconnect(self) -> None:
"""Close the session and terminate the server subprocess."""
if self._exit_stack:
await self._exit_stack.aclose()
self._exit_stack = None
self.session = None
self._tools_cache = None
logger.info("MCP session closed")
The AsyncExitStack pattern ensures that all resources — the subprocess, the transport streams, and the session — are cleaned up in the correct order when disconnect is called. This is important because MCP sessions hold open pipes to the server process, and failing to close them leaves orphaned processes.
Step 2: Discovering Tools
Once connected, the first thing a good client does is discover what tools the server offers. We cache the result so repeated calls don't incur unnecessary round trips.
async def list_tools(self, refresh: bool = False) -> list[dict]:
"""Return the list of tools exposed by the server."""
if self.session is None:
raise RuntimeError("Client is not connected. Call connect() first.")
if self._tools_cache is not None and not refresh:
return self._tools_cache
result = await asyncio.wait_for(
self.session.list_tools(),
timeout=self.timeout
)
self._tools_cache = [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema
}
for tool in result.tools
]
logger.info(f"Discovered {len(self._tools_cache)} tools")
return self._tools_cache
Step 3: Invoking Tools
The core of any MCP client is tool invocation. We wrap the raw call_tool method with timeout handling, response parsing, and structured error reporting.
async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
"""Invoke a tool by name with the given arguments and return text output."""
if self.session is None:
raise RuntimeError("Client is not connected. Call connect() first.")
logger.info(f"Calling tool '{name}' with arguments: {arguments}")
try:
result = await asyncio.wait_for(
self.session.call_tool(name, arguments),
timeout=self.timeout
)
except asyncio.TimeoutError:
raise TimeoutError(
f"Tool '{name}' did not respond within {self.timeout}s"
)
# Extract text from the content blocks
text_parts = []
for block in result.content:
if hasattr(block, "text"):
text_parts.append(block.text)
else:
text_parts.append(json.dumps(block.dict()))
if result.isError:
raise RuntimeError(
f"Tool '{name}' returned an error: {' '.join(text_parts)}"
)
return "\n".join(text_parts)
Notice that we check result.isError. MCP distinguishes between transport-level errors (which raise exceptions) and tool-level errors (which return successfully but with an error flag). A robust client handles both.
Step 4: High-Level Convenience Methods
While call_tool is generic, it's helpful to provide typed convenience methods for the specific operations our file server supports. This makes the client ergonomic to use in application code.
async def read_file(self, filename: str) -> str:
"""Read a file from the sandbox."""
return await self.call_tool("read_file", {"filename": filename})
async def list_directory(self, subdir: str = "") -> list[str]:
"""List entries in a sandbox directory."""
raw = await self.call_tool("list_directory", {"subdir": subdir})
return json.loads(raw)
async def write_file(self, filename: str, content: str) -> str:
"""Write content to a file in the sandbox."""
return await self.call_tool("write_file", {
"filename": filename,
"content": content
})
Step 5: Putting It All Together
Now let's write a main function that demonstrates the full workflow: connecting, listing tools, reading a file, writing a new file, and listing the directory.
async def main():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
client = FileMCPClient(
server_command=["python", "server.py"],
timeout=15.0
)
try:
await client.connect()
# Discover tools
tools = await client.list_tools()
print("\nAvailable tools:")
for t in tools:
print(f" - {t['name']}: {t['description']}")
# List the sandbox directory
print("\nDirectory listing:")
entries = await client.list_directory()
for entry in entries:
print(f" {entry}")
# Read the sample file
print("\nReading sample.txt:")
content = await client.read_file("sample.txt")
print(content)
# Write a new file
print("\nWriting new_file.txt:")
result = await client.write_file("new_file.txt", "This file was created by the MCP client!")
print(result)
# List again to confirm
print("\nUpdated directory listing:")
entries = await client.list_directory()
for entry in entries:
print(f" {entry}")
except Exception as e:
print(f"\nError: {e}")
finally:
await client.disconnect()
if __name__ == "__main__":
asyncio.run(main())
Run the client from the project directory:
python client.py
You should see output showing the three discovered tools, the initial directory listing, the contents of sample.txt, a confirmation that new_file.txt was written, and the updated directory listing with the new file present.
Adding a Permission Layer
One of the main reasons to build a custom client is to insert application-specific logic between the user (or AI agent) and the actual tool execution. A permission layer is a perfect example. Let's extend our client to require explicit approval before any write operation.
async def call_tool_with_permission(
self,
name: str,
arguments: dict[str, Any],
require_confirmation: bool = False
) -> str:
"""Call a tool, optionally prompting for user confirmation."""
if require_confirmation:
print(f"\n[Permission required] Tool: {name}")
print(f"Arguments: {json.dumps(arguments, indent=2)}")
response = input("Allow this operation? (y/n): ").strip().lower()
if response != "y":
raise PermissionError(f"User denied execution of tool '{name}'")
return await self.call_tool(name, arguments)
async def write_file(self, filename: str, content: str) -> str:
"""Write content to a file, requiring user confirmation."""
return await self.call_tool_with_permission(
"write_file",
{"filename": filename, "content": content},
require_confirmation=True
)
This pattern is especially valuable when the client is driven by an AI agent. You can classify tools as "safe" (read-only) or "sensitive" (mutating) and only prompt for the latter. In a production system, you might replace the input() call with a UI dialog, an approval API, or a policy engine.
Implementing Retries and Resilience
Real-world MCP servers can fail transiently. A subprocess might crash, a tool might time out, or the server might be temporarily unresponsive. Let's add retry logic to our call_tool method.
async def call_tool(
self,
name: str,
arguments: dict[str, Any],
max_retries: int = 3,
retry_delay: float = 1.0
) -> str:
"""Invoke a tool with retry logic for transient failures."""
if self.session is None:
raise RuntimeError("Client is not connected. Call connect() first.")
last_error: Optional[Exception] = None
for attempt in range(1, max_retries + 1):
try:
logger.info(
f"Calling tool '{name}' (attempt {attempt}/{max_retries})"
)
result = await asyncio.wait_for(
self.session.call_tool(name, arguments),
timeout=self.timeout
)
text_parts = []
for block in result.content:
if hasattr(block, "text"):
text_parts.append(block.text)
else:
text_parts.append(json.dumps(block.dict()))
if result.isError:
raise RuntimeError(
f"Tool '{name}' error: {' '.join(text_parts)}"
)
return "\n".join(text_parts)
except asyncio.TimeoutError:
last_error = TimeoutError(
f"Tool '{name}' timed out after {self.timeout}s"
)
logger.warning(str(last_error))
except RuntimeError as e:
last_error = e
logger.warning(f"Tool error on attempt {attempt}: {e}")
# Don't retry on tool-level errors (the tool ran, it just failed)
raise
except Exception as e:
last_error = e
logger.warning(f"Unexpected error on attempt {attempt}: {e}")
if attempt < max_retries:
await asyncio.sleep(retry_delay * attempt)
raise last_error # type: ignore[misc]
The key design decision here is that tool-level errors (where isError is true) are not retried, because the tool executed successfully — it just reported a logical failure. Only transport-level failures and timeouts trigger a retry. This prevents accidentally writing a file twice when the first write succeeded but the response was lost.
Best Practices for MCP File-Access Clients
1. Always Sandbox File Operations
Never expose arbitrary filesystem access through MCP without a sandbox. Path traversal attacks are trivial to execute and devastating in effect. As shown in our server, resolve all paths against a fixed base directory and reject any that escape it. On the client side, validate filenames before sending them — don't rely solely on the server.
2. Use Timeouts Everywhere
Every await that involves the server should have a timeout. A hung subprocess can freeze your entire application. The asyncio.wait_for wrapper is your friend. Choose timeouts based on the expected operation: reading a small file should be fast (5–10 seconds), while scanning a large directory tree might need more.
3. Log All Tool Invocations
For debugging and auditing, log every tool call with its name, arguments, result, and duration. This is invaluable when something goes wrong, and it's essential if the client is driven by an AI agent whose behavior you need to trace.
import time
# Inside call_tool, before and after the invocation:
start = time.monotonic()
result = await asyncio.wait_for(...)
duration = time.monotonic() - start
logger.info(f"Tool '{name}' completed in {duration:.2f}s")
4. Handle Content Types Explicitly
MCP tool results can contain multiple content blocks of different types — text, images, embedded resources. Don't assume every result is plain text. Inspect the type of each block and handle accordingly. For a file-access client, text is usually sufficient, but if you add binary file support, you'll need to handle base64-encoded image or blob content.
5. Clean Up Reliably
Always use AsyncExitStack or an equivalent context manager pattern to ensure the server subprocess is terminated. If your application crashes without calling disconnect, the subprocess may linger. Consider registering a signal handler or using atexit as a safety net.
6. Validate Tool Schemas Client-Side
Before sending arguments to the server, validate them against the tool's inputSchema. This catches errors early, provides better feedback to the user, and reduces unnecessary round trips. You can use a library like jsonschema for this:
import jsonschema
def validate_arguments(tool: dict, arguments: dict) -> None:
"""Validate arguments against a tool's input schema."""
schema = tool.get("inputSchema", {})
jsonschema.validate(instance=arguments, schema=schema)
7. Separate Read and Write Permissions
Design your client so that read operations and write operations go through different code paths. This makes it easy to apply stricter confirmation, logging, or rate limiting to writes without burdening reads. It also aligns with the principle of least privilege: if an agent only needs to read files, give it a client that has no write methods at all.
Testing the Client
A custom client deserves proper tests. Here's a simple test using pytest and pytest-asyncio that verifies the core workflow:
# test_client.py
import pytest
import os
from client import FileMCPClient
SERVER_CMD = ["python", "server.py"]
SANDBOX = os.path.join(os.path.dirname(__file__), "sandbox")
@pytest.fixture
async def client():
c = FileMCPClient(SERVER_CMD, timeout=10.0)
await c.connect()
yield c
await c.disconnect()
@pytest.mark.asyncio
async def test_list_tools(client):
tools = await client.list_tools()
names = [t["name"] for t in tools]
assert "read_file" in names
assert "write_file" in names
assert "list_directory" in names
@pytest.mark.asyncio
async def test_read_file(client):
content = await client.read_file("sample.txt")
assert "Hello from the local filesystem" in content
@pytest.mark.asyncio
async def test_write_and_read(client):
await client.call_tool("write_file", {
"filename": "test_output.txt",
"content": "Test content"
})
content = await client.read_file("test_output.txt")
assert content == "Test content"
# Clean up
os.remove(os.path.join(SANDBOX, "test_output.txt"))
@pytest.mark.asyncio
async def test_path_traversal_blocked(client):
with pytest.raises(Exception):
await client.read_file("../../etc/passwd")
Install the test dependencies and run:
pip install pytest pytest-asyncio jsonschema
pytest -v
Extending the Client
Once you have a working file-access client, there are many directions to extend it:
- Resource subscriptions: MCP supports subscribing to resource changes. You could watch a directory and notify your application when files are added or modified.
- Multiple server connections: Connect to several MCP servers simultaneously — one for files, one for a database, one for web search — and route tool calls to the appropriate one.
- Streaming responses: For large file reads, use MCP's streaming capabilities to process content incrementally rather than buffering it all in memory.
- AI agent integration: Wrap the client in an agent loop that uses an LLM to decide which tools to call based on user requests, passing the tool schemas as function definitions.
Here's a sketch of multi-server support using a registry pattern:
class MultiServerMCPClient:
def __init__(self):
self._clients: dict[str, FileMCPClient] = {}
async def add_server(self, name: str, command: list[str]) -> None:
client = FileMCPClient(command)
await client.connect()
self._clients[name] = client
async def call_tool(self, server_name: str, tool_name: str, args: dict) -> str:
if server_name not in self._clients:
raise KeyError(f"Unknown server: {server_name}")
return await self._clients[server_name].call_tool(tool_name, args)
async def disconnect_all(self) -> None:
for client in self._clients.values():
await client.disconnect()
self._clients.clear()
Conclusion
Building a custom MCP client for local file access gives you deep control over how your application interacts with the filesystem through the Model Context Protocol. By handling connection lifecycle, tool discovery, invocation, error recovery, and permissions yourself, you can tailor the client precisely to your application's needs — whether that means strict confirmation flows for writes, detailed audit logging, or integration with an AI agent's decision loop. The patterns we've covered here — sandboxing, timeouts, retries, schema validation, and clean resource management — form a solid foundation that scales from simple scripts to production systems. As the MCP ecosystem grows, a well-built client will let you adopt new servers and capabilities without rewriting your application's core logic, making the upfront investment in a custom client well worth the effort.