	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP File System Server with ADK

MCP File System Server with ADK

Author: Venkata Sudhakar

ADK agents that need to read reports, write summaries, or search through documents benefit from a dedicated file system MCP server. Rather than giving the agent direct file system access, the MCP server acts as a controlled gateway - it sandboxes all operations to a specific directory, validates paths, and exposes only the operations the agent actually needs. This separation makes the system auditable and prevents path traversal attacks.

The file system MCP server exposes tools for listing files, reading content, writing new files, and searching file contents by keyword. All paths are validated against a base directory before any operation executes. The ADK agent uses these tools to work with documents as naturally as if it had direct file access, but all I/O flows through the controlled MCP interface.

The below example shows a ShopMax India MCP server managing a reports directory, with an ADK agent that reads and summarises monthly sales reports.

# shopmax_fs_server.py - File system MCP server
import os
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("shopmax-filesystem")
BASE_DIR = Path("/tmp/shopmax_reports")  # sandboxed directory
BASE_DIR.mkdir(exist_ok=True)

# Seed some demo report files
(BASE_DIR / "march_2026_sales.txt").write_text(
    "ShopMax India - March 2026 Sales Report\n"
    "Total Revenue: Rs 18,42,000\n"
    "Top Product: Laptop (142 units)\n"
    "Top City: Mumbai (Rs 4,20,000)\n"
    "Returns: 12 units (0.8%)\n"
)
(BASE_DIR / "april_2026_sales.txt").write_text(
    "ShopMax India - April 2026 Sales Report\n"
    "Total Revenue: Rs 21,15,000\n"
    "Top Product: TV (198 units)\n"
    "Top City: Delhi (Rs 5,10,000)\n"
    "Returns: 8 units (0.4%)\n"
)

def safe_path(filename: str) -> Path:
    # Prevent path traversal attacks
    path = (BASE_DIR / filename).resolve()
    if not str(path).startswith(str(BASE_DIR)):
        raise ValueError("Access denied: path outside sandbox")
    return path

@mcp.tool()
def list_files() -> list:
    # List all files in the reports directory
    return [f.name for f in BASE_DIR.iterdir() if f.is_file()]

@mcp.tool()
def read_file(filename: str) -> dict:
    # Read content of a specific report file
    path = safe_path(filename)
    if not path.exists():
        return {"error": "File not found"}
    return {"filename": filename, "content": path.read_text()}

@mcp.tool()
def write_file(filename: str, content: str) -> dict:
    # Write content to a file in the reports directory
    path = safe_path(filename)
    path.write_text(content)
    return {"written": True, "filename": filename, "bytes": len(content)}

@mcp.tool()
def search_files(keyword: str) -> list:
    # Search for a keyword across all report files
    matches = []
    for f in BASE_DIR.iterdir():
        if f.is_file() and keyword.lower() in f.read_text().lower():
            matches.append(f.name)
    return matches

if __name__ == "__main__":
    mcp.run(transport="stdio")

Now connect an ADK agent to work with the file system via MCP,

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

async def main():
    mcp_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python", args=["shopmax_fs_server.py"]
        )
    )

agent = LlmAgent(
        name="report_agent",
        model="gemini-2.0-flash",
        tools=mcp_tools,
        instruction=(
            "You are a ShopMax India report analyst. "
            "Use file tools to list, read and summarise sales reports. "
            "After summarising, save the summary to a new file called summary.txt."
        )
    )

session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="report_agent", session_service=session_service)
    session = session_service.create_session(app_name="report_agent", user_id="mgr1")

query = "List available reports, read them all, and write a combined summary to summary.txt."
    msg = types.Content(role="user", parts=[types.Part(text=query)])
    async for event in runner.run_async(user_id="mgr1", session_id=session.id, new_message=msg):
        if event.is_final_response():
            print(event.content.parts[0].text)

await exit_stack.aclose()

asyncio.run(main())

It gives the following output,

I found 2 reports: march_2026_sales.txt and april_2026_sales.txt.

Combined Summary:
- March 2026: Revenue Rs 18,42,000 | Top: Laptop | Best City: Mumbai
- April 2026: Revenue Rs 21,15,000 | Top: TV    | Best City: Delhi
- Growth: +14.8% month-on-month
- Returns improved from 0.8% to 0.4%

Summary saved to summary.txt.

The sandbox validation in safe_path() is critical - always resolve the full path and check it starts with BASE_DIR before any file operation. In production, run the MCP file server with a dedicated OS user that has read/write access only to the reports directory. For large document collections, add a vector search tool alongside the keyword search to let the agent find semantically similar content across hundreds of files. Use write_file for agent-generated summaries, audit reports, and automated responses that feed back into downstream workflows.

Send your comments, suggestions or queries regarding this site to [email protected].