MCP Tools: Developer Guide to Building Model-Controlled Tools
TL;DR — MCP tools are model-controlled functions AI invokes to interact with external systems. MCP Spec: "Tools enable models to interact with external systems — querying databases, calling APIs, performing computations. Model-controlled: language model decides when and how to use them." MCP Info: "Testing strategy: functional (valid/invalid inputs), security (auth, sanitization, rate limiting), performance (behavior under load)." DeepWiki: "@mcp.tool() decorator with ToolManager. Managers handle storage and lookup." DEV Community: "@mcp.tool() auto-generates schema from type hints. Docstring is what AI reads to decide tool calls — write clearly." WebFuse: "MCP cheat sheet: architecture, primitives, transports, SDK quick starts, security best practices." Learn more with MCP guide, build MCP server, build MCP client, and MCP transports.
MCP Spec defines tools: "The Model Context Protocol allows servers to expose tools that can be invoked by language models. Tools enable models to interact with external systems, such as querying databases, calling APIs, or performing computations. Tools in MCP are designed to be model-controlled, meaning that the language model decides when and how to use them."
MCP Info describes testing: "A comprehensive testing strategy for MCP tools should cover: Functional testing — verify tools execute correctly with valid inputs and handle invalid inputs appropriately. Security testing — validate authentication, authorization, input sanitization, and rate limiting. Performance testing — check behavior under load."
MCP Tools Architecture
registers function as tool"] TypeHints["Type Hints
str, int, float, bool, list
auto-generates JSON schema"] Docstring["Docstring
AI reads to decide
whether to call tool"] Defaults["Default Values
optional parameters
with defaults"] end subgraph Schema["Auto-Generated Schema"] Name["tool name
from function name"] Description["description
from docstring"] InputSchema["inputSchema
JSON Schema from type hints"] Output["output
return type annotation"] end subgraph Invocation["Tool Invocation"] Discover["tools/list
client discovers available tools"] Call["tools/call
client invokes tool with args"] Execute["Execute Function
server runs Python function"] Result["Content Result
text, image, or resource"] Error["Error Response
code and message"] end subgraph Features["Advanced Features"] Progress["Progress Notifications
ctx.report_progress()
user sees tool working"] Cancellation["Cancellation
ctx.is_cancelled()
user aborts long ops"] Logging["Logging
ctx.info(), ctx.warning()
structured logs"] Context["Context Object
access to request info
and MCP features"] end subgraph Testing["Testing Strategy"] Functional["Functional Testing
valid and invalid inputs"] Security["Security Testing
auth, sanitization, rate limit"] Performance["Performance Testing
behavior under load"] end Definition --> Schema Schema --> Invocation Features --> Invocation Invocation --> Testing
Tool Properties
| Property | Source | Required | Example |
|---|---|---|---|
| name | Function name | Yes | get_repo_info |
| description | Docstring | Yes | Fetch metadata for a GitHub repo |
| inputSchema | Type hints | Yes | {"type": "object", "properties": {"query": {"type": "string"}}} |
| return type | Return annotation | No | str, dict, list |
Content Types in Tool Results
| Content Type | Use Case | Format |
|---|---|---|
| Text | Most common — strings, JSON, formatted output | {"type": "text", "text": "result"} |
| Image | Screenshots, charts, visual data | {"type": "image", "data": "base64", "mimeType": "image/png"} |
| Resource | Links to MCP resources | {"type": "resource", "resource": {"uri": "docs://x"}} |
Implementation
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class ContentType(Enum):
TEXT = "text"
IMAGE = "image"
RESOURCE = "resource"
@dataclass
class MCPToolsGuide:
"""Developer guide for building MCP tools."""
def get_basic_tool(self) -> str:
"""Basic MCP tool definition."""
return (
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP('my-server')\n"
"\n"
"@mcp.tool()\n"
"async def get_data(\n"
" query: str, limit: int = 10\n"
") -> str:\n"
" '''Get data for a search query.\n"
"\n"
" Args:\n"
" query: The search query string\n"
" limit: Maximum results to return\n"
" '''\n"
" # Tool implementation\n"
" results = await search(query, limit)\n"
" return format_results(results)"
)
def get_tool_with_validation(self) -> str:
"""Tool with input validation using Pydantic."""
return (
"from pydantic import BaseModel, Field\n"
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP('api-server')\n"
"\n"
"class SearchParams(BaseModel):\n"
" query: str = Field(\n"
" ...,\n"
" min_length=1,\n"
" max_length=500,\n"
" description='Search query'\n"
" )\n"
" limit: int = Field(\n"
" default=10,\n"
" ge=1,\n"
" le=100,\n"
" description='Max results'\n"
" )\n"
"\n"
"@mcp.tool()\n"
"async def search_api(\n"
" query: str, limit: int = 10\n"
") -> dict:\n"
" '''Search the API with validation.\n"
"\n"
" Args:\n"
" query: Search query (1-500 chars)\n"
" limit: Max results (1-100)\n"
" '''\n"
" # Validate with Pydantic\n"
" params = SearchParams(\n"
" query=query, limit=limit)\n"
" \n"
" # Check API key\n"
" import os\n"
" api_key = os.environ.get('API_KEY')\n"
" if not api_key:\n"
" return {\n"
" 'error': 'API key not configured',\n"
" 'details': 'Set API_KEY env var'\n"
" }\n"
" \n"
" # Make API call\n"
" import httpx\n"
" async with httpx.AsyncClient() as client:\n"
" resp = await client.get(\n"
" 'https://api.example.com/search',\n"
" params={'q': params.query,\n"
" 'limit': params.limit},\n"
" headers={'Authorization':\n"
" f'Bearer {api_key}'},\n"
" timeout=10.0,\n"
" )\n"
" \n"
" if resp.status_code != 200:\n"
" return {\n"
" 'error': 'API request failed',\n"
" 'status': resp.status_code,\n"
" 'details': resp.text\n"
" }\n"
" \n"
" return resp.json()"
)
def get_tool_with_progress(self) -> str:
"""Tool with progress notifications and cancellation."""
return (
"from mcp.server.fastmcp import FastMCP, Context\n"
"\n"
"mcp = FastMCP('batch-server')\n"
"\n"
"@mcp.tool()\n"
"async def batch_process(\n"
" items: list[str],\n"
" ctx: Context\n"
") -> str:\n"
" '''Process a batch of items.\n"
"\n"
" Args:\n"
" items: List of items to process\n"
" '''\n"
" total = len(items)\n"
" results = []\n"
" \n"
" for i, item in enumerate(items):\n"
" # Check for cancellation\n"
" if ctx.is_cancelled():\n"
" await ctx.info(\n"
" 'Batch cancelled by user')\n"
" return f'Cancelled at item {i}'\n"
" \n"
" # Report progress\n"
" await ctx.report_progress(\n"
" i + 1, total)\n"
" \n"
" # Log progress\n"
" await ctx.info(\n"
" f'Processing {i+1}/{total}: '\n"
" f'{item}')\n"
" \n"
" # Process item\n"
" result = await process_item(item)\n"
" results.append(result)\n"
" \n"
" return f'Processed {total} items'"
)
def get_tool_error_handling(self) -> str:
"""Tool with comprehensive error handling."""
return (
"import httpx\n"
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP('robust-server')\n"
"\n"
"@mcp.tool()\n"
"async def fetch_url(url: str) -> str:\n"
" '''Fetch content from a URL.\n"
"\n"
" Args:\n"
" url: The URL to fetch\n"
" '''\n"
" # Validate URL\n"
" if not url.startswith(('http://',\n"
" 'https://')):\n"
" return {\n"
" 'error': 'Invalid URL',\n"
" 'details': 'URL must start '\n"
" 'with http:// or https://'\n"
" }\n"
" \n"
" try:\n"
" async with httpx.AsyncClient() as client:\n"
" resp = await client.get(\n"
" url, timeout=10.0,\n"
" follow_redirects=True)\n"
" resp.raise_for_status()\n"
" return resp.text[:5000]\n"
" \n"
" except httpx.TimeoutException:\n"
" return {\n"
" 'error': 'Request timed out',\n"
" 'url': url\n"
" }\n"
" except httpx.HTTPStatusError as e:\n"
" return {\n"
" 'error': 'HTTP error',\n"
" 'status': e.response.status_code,\n"
" 'url': url\n"
" }\n"
" except httpx.RequestError as e:\n"
" return {\n"
" 'error': 'Request failed',\n"
" 'details': str(e)\n"
" }\n"
" except Exception as e:\n"
" return {\n"
" 'error': 'Unexpected error',\n"
" 'details': str(e)\n"
" }"
)
def get_tool_returning_image(self) -> str:
"""Tool returning image content."""
return (
"import base64\n"
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP('chart-server')\n"
"\n"
"@mcp.tool()\n"
"async def generate_chart(\n"
" data: list[float],\n"
" title: str = 'Chart'\n"
") -> dict:\n"
" '''Generate a bar chart image.\n"
"\n"
" Args:\n"
" data: List of numeric values\n"
" title: Chart title\n"
" '''\n"
" # Generate chart (e.g., matplotlib)\n"
" import matplotlib.pyplot as plt\n"
" import io\n"
" \n"
" plt.bar(range(len(data)), data)\n"
" plt.title(title)\n"
" \n"
" buf = io.BytesIO()\n"
" plt.savefig(buf, format='png')\n"
" buf.seek(0)\n"
" plt.close()\n"
" \n"
" image_b64 = base64\\\n"
" .b64encode(buf.read())\\\n"
" .decode('utf-8')\n"
" \n"
" return {\n"
" 'type': 'image',\n"
" 'data': image_b64,\n"
" 'mimeType': 'image/png'\n"
" }"
)
def get_testing_strategy(self) -> dict:
"""Testing strategy for MCP tools."""
return {
"functional_testing": {
"valid_inputs": [
"Test with correct parameter types",
"Test with default values",
"Test with optional parameters omitted",
"Test with edge case values",
"Verify correct output format",
],
"invalid_inputs": [
"Test with wrong parameter types",
"Test with missing required parameters",
"Test with out-of-range values",
"Test with empty strings",
"Test with None values",
"Verify error messages are helpful",
],
},
"security_testing": {
"authentication": [
"Test with missing API keys",
"Test with invalid API keys",
"Test with expired tokens",
"Verify auth failures are handled",
],
"input_sanitization": [
"Test with SQL injection attempts",
"Test with XSS payloads",
"Test with path traversal attempts",
"Test with command injection",
"Verify inputs are sanitized",
],
"rate_limiting": [
"Test rapid successive calls",
"Verify rate limits are enforced",
"Test concurrent tool calls",
],
},
"performance_testing": {
"load": [
"Test with large input data",
"Test with many concurrent calls",
"Test timeout behavior",
"Test memory usage",
],
"monitoring": [
"Track response times",
"Track error rates",
"Track tool call frequency",
"Monitor resource usage",
],
},
}
def get_best_practices(self) -> dict:
"""MCP tool best practices."""
return {
"naming": (
"Use clear, descriptive names. "
"AI uses the name to decide if "
"the tool is relevant."
),
"docstrings": (
"Write detailed docstrings. "
"AI reads the docstring to decide "
"whether to call the tool. Include "
"parameter descriptions and examples."
),
"type_hints": (
"Use specific type hints for all "
"parameters. FastMCP auto-generates "
"JSON schema from them."
),
"defaults": (
"Provide default values for optional "
"parameters to make tools more "
"flexible."
),
"errors": (
"Return structured error objects "
"instead of raising exceptions when "
"possible. Include helpful error "
"messages."
),
"async": (
"Use async functions for I/O "
"operations (API calls, database "
"queries, file operations)."
),
"security": (
"Validate all inputs. Check API "
"keys before external calls. "
"Sanitize inputs. Use least-"
"privilege credentials."
),
"progress": (
"Use Context for long-running "
"operations. Report progress. "
"Check for cancellation."
),
}
MCP Tools Checklist
- [ ] MCP tools are model-controlled — language model decides when and how to use them
- [ ] Tools enable models to interact with external systems: databases, APIs, computations
- [ ] Use @mcp.tool() decorator to register a Python function as an MCP tool
- [ ] FastMCP auto-generates JSON schema from type hints and docstring
- [ ] Tool properties: name (from function), description (from docstring), inputSchema (from type hints)
- [ ] Write clear, detailed docstrings — AI reads them to decide whether to call the tool
- [ ] Use specific type hints: str, int, float, bool, list, dict, Optional
- [ ] Provide default values for optional parameters
- [ ] Use async functions for I/O operations (API calls, database queries, file operations)
- [ ] ToolManager handles internal storage and lookup of registered tools
- [ ] Server advertises tools via tools/list — client discovers available tools
- [ ] Client invokes tools via tools/call with arguments
- [ ] Tool result is content list: text, image, or resource
- [ ] Error responses include code and message
- [ ] Return structured error objects instead of raising exceptions when possible
- [ ] Include helpful error messages with details
- [ ] Validate inputs with type hints and Pydantic models
- [ ] Check API keys and tokens before making external calls
- [ ] Sanitize inputs to prevent injection attacks (SQL, XSS, path traversal)
- [ ] Use least-privilege API keys and tokens
- [ ] Rate limit tool calls to prevent abuse
- [ ] Content types: text (most common), image (base64 + mimeType), resource (URI link)
- [ ] Use Context object for progress notifications: ctx.report_progress(current, total)
- [ ] Use Context for logging: ctx.info(), ctx.warning(), ctx.error()
- [ ] Check for cancellation: ctx.is_cancelled() — user can abort long operations
- [ ] Clean up resources on cancellation
- [ ] Progress notifications let user see tool is working during long operations
- [ ] Testing strategy: functional, security, performance
- [ ] Functional testing: valid inputs (correct types, defaults, edge cases, output format)
- [ ] Functional testing: invalid inputs (wrong types, missing params, out-of-range, None)
- [ ] Security testing: authentication (missing/invalid/expired keys, auth failure handling)
- [ ] Security testing: input sanitization (SQL injection, XSS, path traversal, command injection)
- [ ] Security testing: rate limiting (rapid calls, concurrent calls, limit enforcement)
- [ ] Performance testing: load (large inputs, concurrent calls, timeout, memory)
- [ ] Performance testing: monitoring (response times, error rates, call frequency, resources)
- [ ] Use httpx for async HTTP calls in tool implementations
- [ ] Set timeouts on all external calls (e.g., timeout=10.0)
- [ ] Handle common exceptions: TimeoutException, HTTPStatusError, RequestError
- [ ] Best practice: clear naming — AI uses name to decide tool relevance
- [ ] Best practice: detailed docstrings with parameter descriptions
- [ ] Best practice: specific type hints for auto-generated schema
- [ ] Best practice: defaults for optional parameters
- [ ] Best practice: structured error objects with helpful messages
- [ ] Best practice: async for I/O operations
- [ ] Best practice: validate all inputs, check auth, sanitize, least-privilege
- [ ] Best practice: Context for long-running operations (progress, cancellation, logging)
- [ ] Test with MCP Inspector for integration testing
- [ ] Unit test each tool function independently
- [ ] Read MCP developer guide for protocol
- [ ] Read build MCP server for server creation
- [ ] Read build MCP client for client implementation
- [ ] Read MCP transports for transport details
- [ ] Read Claude Desktop MCP for desktop integration
- [ ] Test: tool executes correctly with valid inputs and returns expected output
- [ ] Test: tool handles invalid inputs with helpful error messages
- [ ] Test: tool checks authentication before external calls
- [ ] Test: tool sanitizes inputs against injection attacks
- [ ] Test: progress notifications work during long operations
- [ ] Test: cancellation aborts long operations cleanly
- [ ] Document tool schemas, error handling, security measures, and testing strategy
FAQ
What are MCP tools and how do they work?
MCP tools are model-controlled functions that language models can invoke to interact with external systems like databases, APIs, and file systems. MCP Spec: "The Model Context Protocol allows servers to expose tools that can be invoked by language models. Tools enable models to interact with external systems, such as querying databases, calling APIs, or performing computations. Tools in MCP are designed to be model-controlled, meaning that the language model decides when and how to use them." MCP Info: "A comprehensive testing strategy for MCP tools should cover: functional testing (valid and invalid inputs), security testing (auth, authorization, input sanitization, rate limiting), performance testing." Tools have: name, description, inputSchema (JSON Schema). Server advertises tools via tools/list. Client invokes via tools/call. Result is content list (text, image, resource).
How do you define an MCP tool with the @mcp.tool() decorator?
Use @mcp.tool() decorator on a Python function. FastMCP auto-generates JSON schema from type hints and docstring. DeepWiki: "MCPServer provides @mcp.tool() decorator with ToolManager. Managers handle internal storage and lookup logic." DEV Community: "@mcp.tool() decorator auto-generates a schema from your type hints. The docstring is what the AI reads to decide whether to call this tool — write it clearly." MCP Docs: "FastMCP class uses Python type hints and docstrings to automatically generate tool definitions." Definition: @mcp.tool() async def get_data(query: str, limit: int = 10) -> str: '''Get data for a query. Args: query: The search query. limit: Max results to return. ''' return results. Schema auto-generated from: function name, parameter type hints, default values, docstring.
How do you handle errors and validation in MCP tools?
Return error content in the tool result, validate inputs with type hints and Pydantic, and use structured error responses. MCP Spec: "Tools/call response contains either content (success) or error (failure). Error responses include code and message." MCP Info: "Functional testing: verify tools execute correctly with valid inputs and handle invalid inputs appropriately. Security testing: validate authentication, authorization, input sanitization, rate limiting." Error handling: (1) Type hints provide basic validation. (2) Use Pydantic models for complex validation. (3) Return error as content: return {'error': 'Invalid input', 'details': '...'}. (4) Raise exceptions for unexpected errors — FastMCP catches and returns as error response. (5) Validate API keys and tokens before making external calls. (6) Sanitize inputs to prevent injection attacks. (7) Rate limit tool calls.
How do you implement progress notifications and cancellation in MCP tools?
Use the Context object to send progress notifications and check for cancellation during long-running tool operations. MCP Spec: "Utilities include progress tracking, cancellation, logging." MCP Info: "Server config: timeout: 30, max_connections: 100. Monitoring: metrics_enabled." Progress: from mcp.server.fastmcp import Context. @mcp.tool() async def long_task(query: str, ctx: Context) -> str: await ctx.report_progress(50, 100) # 50% done. await ctx.info('Processing...') # log. if ctx.is_cancelled(): return 'Cancelled'. Cancellation: client sends cancellation notification. Server checks ctx.is_cancelled() periodically. Clean up resources on cancellation. Progress notifications let user see tool is working. Cancellation lets user abort long operations.
What testing strategy should you use for MCP tools?
Test functional (valid/invalid inputs), security (auth, sanitization, rate limiting), and performance (behavior under load) aspects of every MCP tool. MCP Info: "A comprehensive testing strategy for MCP tools should cover: Functional testing — verify tools execute correctly with valid inputs and handle invalid inputs appropriately. Security testing — validate authentication, authorization, input sanitization, and rate limiting. Performance testing — check behavior under load." WebFuse: "MCP cheat sheet: architecture, primitives, transports, SDK quick starts, security best practices, and full examples." Testing: (1) Unit test each tool function independently. (2) Test with valid inputs — verify correct output. (3) Test with invalid inputs — verify error handling. (4) Test with missing API keys — verify auth failure. (5) Test with malformed inputs — verify sanitization. (6) Test rate limiting — verify limits enforced. (7) Test with MCP Inspector for integration. (8) Test concurrent tool calls. (9) Test timeout behavior. (10) Test cancellation during long operations.
Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →