Model Context Protocol (MCP) Complete Guide: AI Tool Standardization and Claude Integration

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/tmp/example.txt"
    }
  }
}

[Host: Claude Desktop]
    └── [Client] ──── stdio/SSE ──── [MCP Server: filesystem]
    └── [Client] ──── stdio/SSE ──── [MCP Server: github]
    └── [Client] ──── stdio/SSE ──── [MCP Server: database]

@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
    return f"Review this {language} code and suggest improvements:\n\n{code}"

pip install fastmcp

from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def read_file(path: str) -> str:
    """Read and return the contents of a file."""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"File saved: {path}"

@mcp.resource("file://{path}")
def get_file_resource(path: str) -> str:
    """Expose a file as a resource via URI."""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

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

import os
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def list_directory(path: str = ".") -> list[str]:
    """Return a list of files in a directory."""
    p = Path(path)
    if not p.is_dir():
        raise ValueError(f"{path} is not a directory")
    return [str(item) for item in p.iterdir()]

@mcp.tool()
def search_files(directory: str, pattern: str) -> list[str]:
    """Search for files matching a pattern in a directory."""
    p = Path(directory)
    return [str(f) for f in p.rglob(pattern)]

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

import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'

const server = new McpServer({
  name: 'database-server',
  version: '1.0.0',
})

// Resource: expose DB records via URI
server.resource(
  'user',
  new ResourceTemplate('db://users/{id}', { list: undefined }),
  async (uri, { id }) => {
    const user = await fetchUserById(id)
    return {
      contents: [
        {
          uri: uri.href,
          text: JSON.stringify(user, null, 2),
          mimeType: 'application/json',
        },
      ],
    }
  }
)

// Tool: execute SQL query
server.tool('query_db', { sql: z.string().describe('SQL query to execute') }, async ({ sql }) => {
  const results = await executeQuery(sql)
  return {
    content: [
      {
        type: 'text',
        text: JSON.stringify(results, null, 2),
      },
    ],
  }
})

async function main() {
  const transport = new StdioServerTransport()
  await server.connect(transport)
}

main()

# SSE server startup example
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["/path/to/filesystem_server.py"],
      "env": {
        "ALLOWED_DIRS": "/home/user/documents,/tmp"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
      }
    },
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/path/to/database.db"]
    }
  }
}

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print("Available tools:", [t.name for t in tools.tools])

            # Call a tool
            result = await session.call_tool(
                "read_file",
                arguments={"path": "/tmp/test.txt"}
            )
            print("Result:", result.content[0].text)

asyncio.run(main())

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def create_mcp_agent():
    model = ChatAnthropic(model="claude-3-5-sonnet-20241022")

    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Convert MCP tools to LangChain tools
            tools = await load_mcp_tools(session)

            # Create a ReAct agent
            agent = create_react_agent(model, tools)

            result = await agent.ainvoke({
                "messages": [{"role": "user", "content": "List the files in /tmp"}]
            })
            return result

asyncio.run(create_mcp_agent())

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("safe-server")

DANGEROUS_PATHS = ["/etc", "/sys", "/proc", "/root"]

@mcp.tool()
def delete_file(path: str, confirmed: bool = False) -> str:
    """
    Delete a file. You must explicitly set confirmed=True before deletion proceeds.

    Args:
        path: Path of the file to delete
        confirmed: Explicitly confirm the deletion intent (default: False)
    """
    import os
    from pathlib import Path

    abs_path = str(Path(path).resolve())

    # Block dangerous paths
    for danger in DANGEROUS_PATHS:
        if abs_path.startswith(danger):
            raise ToolError(f"Cannot delete files under {danger}")

    # If called without confirmation, prompt the user
    if not confirmed:
        return (
            f"Warning: You are about to permanently delete {abs_path}. "
            f"Call again with confirmed=True to proceed."
        )

    os.remove(abs_path)
    return f"Deleted: {abs_path}"

import os
from pathlib import Path

ALLOWED_DIRS = os.getenv("ALLOWED_DIRS", "/tmp").split(",")

def is_path_allowed(path: str) -> bool:
    abs_path = Path(path).resolve()
    return any(
        str(abs_path).startswith(allowed.strip())
        for allowed in ALLOWED_DIRS
    )

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
# Run as a dedicated non-root user
RUN useradd -m mcpuser
USER mcpuser
CMD ["python", "server.py"]

from pydantic import BaseModel, validator

class FileReadParams(BaseModel):
    path: str

    @validator("path")
    def validate_path(cls, v):
        p = Path(v).resolve()
        if ".." in str(p):
            raise ValueError("Path traversal attempt detected")
        if not is_path_allowed(str(p)):
            raise ValueError("Path not in allowed list")
        return str(p)