Creating Custom Tools & Resources for MCP Servers

Custom tools and resources are the core of what makes an MCP server useful. Tools give AI models the ability to perform actions -- querying databases, calling APIs, transforming data. Resources give applications access to data -- file contents, configuration, schema information. Together, they define what your MCP server can do.

This guide covers everything you need to design, implement, and optimize custom tools and resources: JSON Schema definitions, advanced input patterns, error handling, pagination, streaming, and prompt templates that tie everything together.

If you have not built an MCP server yet, start with our Python tutorial or TypeScript guide first, then return here to deepen your knowledge.

Understanding the Three Building Blocks

MCP servers expose three types of capabilities, each with a different control model:

For a detailed conceptual overview, see MCP Core Building Blocks. This guide focuses on the practical implementation.

Designing Effective Tool Schemas

Every MCP tool is defined by three things: a name, a description, and an input schema. The AI model uses all three to decide when and how to call your tool.

Anatomy of a Tool Definition

{
  name: "search_documents",
  description: "Search through documents by keyword, returning matching results with relevance scores. Supports filtering by date range and document type.",
  inputSchema: {
    type: "object",
    properties: {
      query: {
        type: "string",
        description: "Search keywords or phrase"
      },
      doc_type: {
        type: "string",
        enum: ["pdf", "markdown", "text", "all"],
        description: "Filter by document type. Defaults to 'all'."
      },
      max_results: {
        type: "number",
        description: "Maximum number of results to return (1-100). Defaults to 10."
      },
      date_after: {
        type: "string",
        description: "Only return documents modified after this date (ISO 8601 format, e.g., '2025-01-01')"
      }
    },
    required: ["query"]
  }
}

Tool Naming Best Practices

Good tool names are concise, action-oriented, and follow consistent conventions:

Use snake_case for tool names. This is the convention across the MCP ecosystem and both official SDKs.

Writing Descriptions That Guide the AI

The tool description is the most important factor in whether the AI model uses your tool correctly. Write descriptions that answer three questions:

1. What does it do? (first sentence)
2. When should the model use it? (key use cases)
3. What should the model know? (limitations, side effects)

@mcp.tool()
async def execute_sql(query: str, database: str = "main") -> str:
    """Execute a read-only SQL query against the specified database.

    Use this tool when the user asks questions about data that can be
    answered with SQL. Only SELECT queries are allowed — INSERT, UPDATE,
    DELETE, and DDL statements will be rejected.

    The 'main' database contains user and order tables.
    The 'analytics' database contains event and metric tables.

    Args:
        query: A valid SQL SELECT statement
        database: Database to query — either 'main' or 'analytics'
    """

Complex Input Schemas

Tools can accept sophisticated input structures using JSON Schema:

Nested Objects

{
  name: "create_chart",
  description: "Generate a chart from data",
  inputSchema: {
    type: "object",
    properties: {
      chart_type: {
        type: "string",
        enum: ["bar", "line", "pie", "scatter"]
      },
      data: {
        type: "object",
        properties: {
          labels: {
            type: "array",
            items: { type: "string" }
          },
          values: {
            type: "array",
            items: { type: "number" }
          }
        },
        required: ["labels", "values"]
      },
      options: {
        type: "object",
        properties: {
          title: { type: "string" },
          width: { type: "number", default: 800 },
          height: { type: "number", default: 600 }
        }
      }
    },
    required: ["chart_type", "data"]
  }
}

Arrays of Objects

from typing import Optional

@mcp.tool()
async def bulk_create_tasks(
    tasks: list[dict],
    project_id: str
) -> str:
    """Create multiple tasks at once in a project.

    Args:
        tasks: List of task objects, each with 'title' (required),
               'description' (optional), and 'priority' (1-5, optional)
        project_id: The project to add tasks to
    """
    created = []
    for task in tasks:
        title = task.get("title")
        if not title:
            continue
        # Create each task...
        created.append(title)
    return f"Created {len(created)} tasks in project {project_id}"

Enum Constraints

Use enums to limit parameter values to a known set:

properties: {
  status: {
    type: "string",
    enum: ["open", "in_progress", "review", "closed"],
    description: "Filter tasks by status"
  },
  sort_by: {
    type: "string",
    enum: ["created_at", "updated_at", "priority", "title"],
    description: "Field to sort results by"
  },
  sort_order: {
    type: "string",
    enum: ["asc", "desc"],
    description: "Sort direction"
  }
}

Enums help the AI model provide valid values and reduce errors.

Implementing Tools in Python

Basic Tool with FastMCP

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My Server")

@mcp.tool()
async def calculate_bmi(
    weight_kg: float,
    height_m: float
) -> str:
    """Calculate Body Mass Index (BMI) from weight and height.

    Args:
        weight_kg: Weight in kilograms
        height_m: Height in meters
    """
    if height_m <= 0:
        return "Error: Height must be a positive number."
    if weight_kg <= 0:
        return "Error: Weight must be a positive number."

    bmi = weight_kg / (height_m ** 2)

    if bmi < 18.5:
        category = "Underweight"
    elif bmi < 25:
        category = "Normal weight"
    elif bmi < 30:
        category = "Overweight"
    else:
        category = "Obese"

    return f"BMI: {bmi:.1f} ({category})"

Tool with Optional Parameters

from typing import Optional

@mcp.tool()
async def search_logs(
    query: str,
    level: str = "all",
    limit: int = 50,
    since: Optional[str] = None
) -> str:
    """Search application logs by keyword.

    Args:
        query: Search string to match against log messages
        level: Filter by log level (debug, info, warn, error, all)
        limit: Maximum number of log entries to return
        since: Only return logs after this ISO timestamp
    """
    # Implementation...

FastMCP automatically:

• Makes query required (no default value)
• Makes level, limit, and since optional with their defaults
• Handles Optional[str] as a nullable string field

Tool Returning Structured Content

from mcp.types import TextContent, ImageContent, EmbeddedResource
import base64

@mcp.tool()
async def generate_qr_code(data: str, size: int = 256) -> list:
    """Generate a QR code image for the given data.

    Args:
        data: The text or URL to encode in the QR code
        size: Image size in pixels (default 256)
    """
    import qrcode
    from io import BytesIO

    qr = qrcode.make(data)
    buffer = BytesIO()
    qr.save(buffer, format="PNG")
    img_base64 = base64.b64encode(buffer.getvalue()).decode()

    return [
        TextContent(type="text", text=f"QR code generated for: {data}"),
        ImageContent(
            type="image",
            data=img_base64,
            mimeType="image/png"
        ),
    ]

Implementing Tools in TypeScript

Tool with Zod Validation

import { z } from "zod";

const SearchSchema = z.object({
  query: z.string().min(1).describe("Search keywords"),
  filters: z.object({
    category: z.enum(["all", "docs", "code", "issues"]).default("all"),
    language: z.string().optional(),
    dateRange: z.object({
      from: z.string().optional(),
      to: z.string().optional(),
    }).optional(),
  }).optional(),
  page: z.number().int().min(1).default(1),
  pageSize: z.number().int().min(1).max(100).default(20),
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "search") {
    const args = SearchSchema.parse(request.params.arguments);

    const results = await performSearch(
      args.query,
      args.filters,
      args.page,
      args.pageSize
    );

    return {
      content: [{
        type: "text",
        text: formatSearchResults(results),
      }],
    };
  }
});

Tool with Side Effects and Confirmation

case "delete_all_completed_tasks": {
  // Tools with destructive side effects should return clear summaries
  const completedTasks = await db.getCompletedTasks();

  if (completedTasks.length === 0) {
    return {
      content: [{
        type: "text",
        text: "No completed tasks to delete.",
      }],
    };
  }

  await db.deleteCompletedTasks();

  return {
    content: [{
      type: "text",
      text: `Deleted ${completedTasks.length} completed tasks:\n` +
        completedTasks.map(t => `- ${t.title}`).join("\n"),
    }],
  };
}

Creating Resources

Resources expose data that applications can read. They are identified by URIs and return content in standard MIME types.

Static Resources (Python)

@mcp.resource("config://app/settings")
def get_app_settings() -> str:
    """Current application settings and configuration."""
    import json
    settings = {
        "version": "2.1.0",
        "environment": "production",
        "features": {
            "dark_mode": True,
            "beta_features": False,
        },
        "limits": {
            "max_upload_mb": 50,
            "rate_limit_per_minute": 100,
        },
    }
    return json.dumps(settings, indent=2)


@mcp.resource("schema://database/users")
def get_users_schema() -> str:
    """Schema definition for the users database table."""
    return """
    Table: users
    ============
    id          INTEGER PRIMARY KEY AUTOINCREMENT
    email       TEXT NOT NULL UNIQUE
    name        TEXT NOT NULL
    role        TEXT DEFAULT 'user' CHECK(role IN ('user','admin','moderator'))
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    updated_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP

    Indexes:
    - idx_users_email (email) UNIQUE
    - idx_users_role (role)
    """

Dynamic Resources with Templates (Python)

@mcp.resource("files://{path}")
def read_file(path: str) -> str:
    """Read the contents of a file at the given path."""
    import os

    # Security: restrict to allowed directory
    base_dir = "/allowed/directory"
    full_path = os.path.normpath(os.path.join(base_dir, path))

    if not full_path.startswith(base_dir):
        raise ValueError("Access denied: path traversal detected")

    with open(full_path, "r") as f:
        return f.read()

Dynamic Resources (TypeScript)

server.setRequestHandler(ListResourcesRequestSchema, async () => {
  // Dynamic: list resources based on current state
  const files = await fs.readdir(docsDirectory);

  return {
    resources: files
      .filter((f) => f.endsWith(".md"))
      .map((filename) => ({
        uri: `docs://${filename}`,
        name: filename.replace(".md", ""),
        description: `Documentation file: ${filename}`,
        mimeType: "text/markdown",
      })),
  };
});

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const uri = request.params.uri;
  const match = uri.match(/^docs:\/\/(.+)$/);

  if (!match) {
    throw new Error(`Unknown resource URI: ${uri}`);
  }

  const filename = match[1];
  const filePath = path.join(docsDirectory, filename);

  // Security check
  if (!filePath.startsWith(docsDirectory)) {
    throw new Error("Access denied");
  }

  const content = await fs.readFile(filePath, "utf-8");

  return {
    contents: [{
      uri,
      mimeType: "text/markdown",
      text: content,
    }],
  };
});

Resource Subscriptions

Clients can subscribe to resource changes for real-time updates:

// Notify clients when a resource changes
async function onFileChanged(filename: string) {
  await server.notification({
    method: "notifications/resources/updated",
    params: { uri: `docs://${filename}` },
  });
}

// Watch for file changes
fs.watch(docsDirectory, (eventType, filename) => {
  if (filename) {
    onFileChanged(filename);
  }
});

Creating Prompt Templates

Prompts structure how users interact with your server's capabilities.

Simple Prompt (Python)

from mcp.server.fastmcp.prompts import base

@mcp.prompt()
def code_review(language: str, code: str) -> list[base.Message]:
    """Review code for bugs, style issues, and improvements."""
    return [
        base.UserMessage(
            content=f"Please review the following {language} code for:\n"
            f"1. Bugs and potential errors\n"
            f"2. Style and readability issues\n"
            f"3. Performance improvements\n"
            f"4. Security concerns\n\n"
            f"```{language}\n{code}\n```"
        )
    ]

Multi-Turn Prompt (TypeScript)

case "data_analysis": {
  const dataset = args?.dataset ?? "default";
  return {
    messages: [
      {
        role: "user",
        content: {
          type: "text",
          text: `I want to analyze the "${dataset}" dataset. First, use the list_columns tool to see what data is available.`,
        },
      },
      {
        role: "assistant",
        content: {
          type: "text",
          text: `I'll start by examining the columns in the "${dataset}" dataset to understand what data we're working with.`,
        },
      },
      {
        role: "user",
        content: {
          type: "text",
          text: "Great. Now provide a statistical summary and identify any interesting patterns or anomalies.",
        },
      },
    ],
  };
}

Prompt with Embedded Resources

case "review_document": {
  const docUri = args?.document_uri;
  return {
    messages: [
      {
        role: "user",
        content: {
          type: "resource",
          resource: {
            uri: docUri,
            mimeType: "text/markdown",
            text: "", // Client will resolve this from the resource
          },
        },
      },
      {
        role: "user",
        content: {
          type: "text",
          text: "Please review this document and provide feedback on clarity, completeness, and accuracy.",
        },
      },
    ],
  };
}

Error Handling Patterns

Robust error handling is critical because AI models need clear feedback to retry or adjust their approach.

The Error Hierarchy

Protocol errors (thrown exceptions)
  └─ Transport failures, malformed requests
  └─ Should NOT be used for tool failures

Tool errors (isError: true in response)
  └─ Tool executed but failed (API down, invalid input, not found)
  └─ AI model receives the error and can try differently

Informational errors (text in response)
  └─ Expected conditions (no results, empty data)
  └─ Part of normal tool output

Python Error Handling

@mcp.tool()
async def fetch_user(user_id: str) -> str:
    """Fetch user details by ID.

    Args:
        user_id: The unique user identifier
    """
    # Input validation
    if not user_id.strip():
        return "Error: user_id cannot be empty."

    try:
        user = await database.get_user(user_id)
    except ConnectionError:
        return "Error: Unable to connect to the database. Please try again later."
    except Exception as e:
        return f"Error: An unexpected error occurred: {str(e)}"

    if user is None:
        return f"No user found with ID '{user_id}'. Please check the ID and try again."

    return (
        f"User: {user['name']}\n"
        f"Email: {user['email']}\n"
        f"Role: {user['role']}\n"
        f"Created: {user['created_at']}"
    )

TypeScript Error Handling

case "fetch_user": {
  const { user_id } = args as { user_id: string };

  if (!user_id?.trim()) {
    return {
      content: [{ type: "text", text: "Error: user_id is required." }],
      isError: true,
    };
  }

  try {
    const user = await db.getUser(user_id);

    if (!user) {
      return {
        content: [{
          type: "text",
          text: `No user found with ID "${user_id}".`,
        }],
      };
    }

    return {
      content: [{
        type: "text",
        text: `Name: ${user.name}\nEmail: ${user.email}\nRole: ${user.role}`,
      }],
    };
  } catch (error) {
    return {
      content: [{
        type: "text",
        text: `Database error: ${error instanceof Error ? error.message : "Unknown"}`,
      }],
      isError: true,
    };
  }
}

Advanced Patterns

Pagination with Cursors

For tools that may return large result sets, implement cursor-based pagination:

import json
import base64

@mcp.tool()
async def list_orders(
    status: str = "all",
    limit: int = 20,
    cursor: str | None = None
) -> str:
    """List orders with pagination.

    Args:
        status: Filter by order status (all, pending, shipped, delivered)
        limit: Number of results per page (max 100)
        cursor: Pagination cursor from a previous response
    """
    limit = min(limit, 100)

    # Decode cursor
    offset = 0
    if cursor:
        try:
            offset = int(base64.b64decode(cursor).decode())
        except Exception:
            return "Error: Invalid pagination cursor."

    # Fetch one extra to check if there's a next page
    orders = await database.list_orders(
        status=status,
        limit=limit + 1,
        offset=offset,
    )

    has_more = len(orders) > limit
    orders = orders[:limit]

    # Build response
    result_lines = [f"Orders (showing {len(orders)} results):"]
    for order in orders:
        result_lines.append(
            f"  [{order['id']}] {order['customer']} - "
            f"${order['total']:.2f} ({order['status']})"
        )

    if has_more:
        next_cursor = base64.b64encode(str(offset + limit).encode()).decode()
        result_lines.append(f"\nMore results available. Use cursor: {next_cursor}")
    else:
        result_lines.append("\nEnd of results.")

    return "\n".join(result_lines)

Long-Running Operations with Progress

For tools that take significant time, use progress notifications:

case "generate_report": {
  const { report_type, date_range } = args;

  // Send progress updates
  await server.notification({
    method: "notifications/progress",
    params: {
      progressToken: request.params._meta?.progressToken,
      progress: 0,
      total: 4,
      message: "Fetching data...",
    },
  });

  const data = await fetchReportData(report_type, date_range);

  await server.notification({
    method: "notifications/progress",
    params: {
      progressToken: request.params._meta?.progressToken,
      progress: 1,
      total: 4,
      message: "Analyzing trends...",
    },
  });

  const analysis = analyzeData(data);

  // ... more steps with progress updates

  return {
    content: [{ type: "text", text: formatReport(analysis) }],
  };
}

Composing Multiple Data Sources

A single tool can orchestrate multiple data sources:

@mcp.tool()
async def customer_360(customer_id: str) -> str:
    """Get a comprehensive 360-degree view of a customer.

    Aggregates data from CRM, order history, support tickets,
    and engagement analytics.

    Args:
        customer_id: The customer ID to look up
    """
    import asyncio

    # Fetch from multiple sources concurrently
    crm_data, orders, tickets, analytics = await asyncio.gather(
        crm_client.get_customer(customer_id),
        orders_db.get_customer_orders(customer_id, limit=10),
        support_client.get_tickets(customer_id, status="open"),
        analytics_client.get_engagement(customer_id),
        return_exceptions=True,
    )

    sections = [f"# Customer 360: {customer_id}\n"]

    # CRM Data
    if isinstance(crm_data, Exception):
        sections.append("## Profile\nUnable to fetch CRM data.\n")
    else:
        sections.append(
            f"## Profile\n"
            f"Name: {crm_data['name']}\n"
            f"Email: {crm_data['email']}\n"
            f"Tier: {crm_data['tier']}\n"
        )

    # Orders
    if isinstance(orders, Exception):
        sections.append("## Recent Orders\nUnable to fetch orders.\n")
    else:
        sections.append(f"## Recent Orders ({len(orders)} shown)\n")
        for order in orders:
            sections.append(
                f"- Order #{order['id']}: ${order['total']:.2f} "
                f"({order['status']})\n"
            )

    # Continue for tickets, analytics...

    return "\n".join(sections)

Tool Design Guidelines

Do

• Keep tools focused -- one clear action per tool
• Use descriptive names -- search_issues_by_label over search
• Document parameters thoroughly -- include examples in descriptions
• Validate inputs -- do not trust the AI model to always send valid data
• Return actionable errors -- tell the model how to fix the problem
• Use enums for parameters with known value sets
• Set sensible defaults -- optional parameters should have good defaults

Do Not

• Expose raw database access without guardrails
• Create tools with too many parameters -- split into multiple tools if needed
• Return massive payloads -- paginate or summarize large results
• Rely on tool ordering -- tools should be independently callable
• Expose internal errors -- translate technical errors into useful messages
• Mix read and write operations in one tool without clear documentation

What to Read Next

• Understand the conceptual foundations: MCP Core Building Blocks
• Build your first server in Python: Build Your First MCP Server in Python
• Build with TypeScript or Go: Building MCP Servers in Node.js, TypeScript & Go
• Test your tools thoroughly: Testing & Debugging MCP Servers
• Browse production server examples: MCP Server Directory

Summary

Creating effective MCP tools and resources comes down to three principles: clear schemas that guide the AI model, robust error handling that enables recovery, and focused design that keeps each tool doing one thing well. The JSON Schema system gives you powerful expressiveness for defining inputs, while the resource and prompt systems let you expose data and workflows in a structured way.

Start with simple tools, test them thoroughly with the MCP Inspector, and iterate based on how the AI model actually uses them. The best MCP servers are not the ones with the most tools, but the ones where every tool is reliable, well-documented, and genuinely useful.