HOW-TO · RAG
How to Create Custom MCP Tools for Your Application
Target environment
Ubuntu 24.04 · Ollama 0.4.x
PREREQUISITES
MCP server set up, Python SDK installed, Python 3.10+
What this does
Custom MCP tools expose your application's APIs, database queries, or business logic as discoverable tools that any MCP-compatible LLM client can invoke. Each tool is an async function with typed parameters.
Steps
- Define a tool with complex parameters. Use typed parameters with defaults and optional fields.
from mcp.server import Server, stdio_server
from typing import Optional
server = Server("custom-app-tools")
@server.tool("search_products")
async def search_products(
query: str,
category: Optional[str] = None,
max_price: Optional[float] = None,
sort_by: str = "relevance",
limit: int = 10
) -> str:
"""Search products in the inventory database."""
# Application logic here
results = [{"id": 1, "name": "Widget", "price": 9.99}]
return str(results)
- Inject application dependencies. Use a factory pattern to create tools with access to shared resources.
from dataclasses import dataclass
@dataclass
class AppContext:
db_connection: str
api_key: str
def create_tools(ctx: AppContext):
server = Server("app-tools")
@server.tool("query_database")
async def query_database(sql: str) -> str:
# ctx.db_connection is available
return f"Executed: {sql} on {ctx.db_connection}"
return server
- Return structured data as JSON strings. MCP tool responses are text; serialize complex data.
import json
@server.tool("get_user_report")
async def get_user_report(user_id: int) -> str:
report = {
"user_id": user_id,
"orders": [{"id": 101, "total": 49.99}],
"metrics": {"visit_count": 42}
}
return json.dumps(report, indent=2)
- Add validation using Pydantic models.
from pydantic import BaseModel, Field
class OrderInput(BaseModel):
user_id: int = Field(gt=0)
items: list[str] = Field(min_length=1)
coupon: Optional[str] = None
@server.tool("place_order")
async def place_order(user_id: int, items: list[str], coupon: str = None) -> str:
validated = OrderInput(user_id=user_id, items=items, coupon=coupon)
return f"Order placed for user {validated.user_id}"
- Handle tool-specific errors. Return meaningful error messages.
@server.tool("delete_user")
async def delete_user(user_id: int) -> str:
if user_id <= 0:
return json.dumps({"error": "Invalid user ID"})
try:
# Perform deletion
return json.dumps({"success": True, "deleted_id": user_id})
except Exception as e:
return json.dumps({"error": str(e)})
Verification
python -c "
from mcp.server import Server
s = Server('custom')
@s.tool('test_tool')
async def test_tool(x: int) -> str:
return str(x * 2)
print(list(s._tool_registry.keys()))
# Expected: ['test_tool']
"
Common failures
- Tool parameters must be JSON-serializable. Python types like
datetimeorsetcause serialization errors. Convert to strings before returning. - Async function not awaited. The SDK expects async functions. Defining
definstead ofasync defcauses registration to fail silently. - Tool name conflicts. Two tools with the same name overwrite each other. Use a prefix like
app_search_productsto namespace. - Version mismatch - The installed package or runtime differs from the command shown; check the version first and rerun the smallest verification command.
- Local environment drift - Another service, virtual environment, model, or path is being used; print the active binary path and configuration before changing the guide steps.
Related guides
- How to Set Up MCP Server with Standard Tools
- How to Connect MCP Client to Remote MCP Server