!pip install -q "mcp[cli]" httpx mistralaiBuild and Deploy MCP Server from Scratch
A hands-on guide to building, testing, and deploying Model Context Protocol servers with Python — tools, resources, prompts, transports, and production hosting
Table of Contents
1. Setup & Installation
Install the required packages for building MCP servers.
2. MCP Architecture Overview
MCP follows a client-server architecture with three participants:
| Participant | Role | Example |
|---|---|---|
| MCP Host | AI application that coordinates MCP clients | Claude Desktop, VS Code Copilot |
| MCP Client | Maintains a connection to one MCP server | Created by the host at runtime |
| MCP Server | Exposes tools, resources, and prompts | Your custom Python server |
Core Primitives
| Primitive | Purpose | Discovery | Execution |
|---|---|---|---|
| Tools | Executable functions the LLM can invoke | tools/list |
tools/call |
| Resources | Read-only data sources providing context | resources/list |
resources/read |
| Prompts | Reusable interaction templates | prompts/list |
prompts/get |
3. Initialize FastMCP Server
FastMCP is the recommended entry point — it wraps JSON-RPC 2.0 protocol handling and provides decorator-based APIs.
from mcp.server.fastmcp import FastMCP
# Create the MCP server instance
mcp = FastMCP("my-knowledge-server")
print(f"Server created: {mcp.name}")4. Define Tools
Tools are functions the LLM can call. The @mcp.tool() decorator registers the function — FastMCP extracts the name, docstring, and type hints to generate the JSON Schema.
import httpx
from typing import Any
@mcp.tool()
async def search_documentation(query: str, max_results: int = 5) -> str:
"""Search the project documentation for relevant articles.
Args:
query: The search query string
max_results: Maximum number of results to return (default: 5)
"""
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.example.com/search",
params={"q": query, "limit": max_results},
timeout=30.0,
)
response.raise_for_status()
results = response.json()
if not results["items"]:
return "No documentation found for this query."
formatted = []
for item in results["items"]:
formatted.append(
f"**{item['title']}**\n{item['snippet']}\nURL: {item['url']}"
)
return "\n---\n".join(formatted)
@mcp.tool()
async def run_sql_query(query: str) -> str:
"""Execute a read-only SQL query against the analytics database.
Args:
query: SQL SELECT query to execute (write operations are blocked)
"""
if not query.strip().upper().startswith("SELECT"):
return "Error: Only SELECT queries are allowed for safety."
# Execute query against your database
# results = await execute_readonly_query(query)
return f"Query executed: {query}"
print("Tools registered: search_documentation, run_sql_query")5. Define Resources
Resources provide read-only context data to the client. Unlike tools, resources are typically user-controlled.
import json
@mcp.resource("config://app-settings")
def get_app_settings() -> str:
"""Return the current application configuration."""
settings = {
"version": "2.1.0",
"environment": "production",
"features": ["search", "analytics", "notifications"],
"rate_limits": {"requests_per_minute": 60},
}
return json.dumps(settings, indent=2)
@mcp.resource("schema://database/{table_name}")
def get_table_schema(table_name: str) -> str:
"""Return the schema definition for a database table.
Args:
table_name: Name of the database table
"""
# Placeholder - load real schemas in production
schemas = {"users": "CREATE TABLE users (id INT, name TEXT, email TEXT)"}
if table_name not in schemas:
return f"Table '{table_name}' not found."
return schemas[table_name]
print("Resources registered: config://app-settings, schema://database/{table_name}")6. Define Prompts
Prompts are reusable interaction templates that structure how users interact with the LLM through your server’s capabilities.
@mcp.prompt()
def debug_error(error_message: str, stack_trace: str = "") -> str:
"""Create a debugging prompt for analyzing errors.
Args:
error_message: The error message to analyze
stack_trace: Optional stack trace for context
"""
context = f"Error: {error_message}"
if stack_trace:
context += f"\n\nStack Trace:\n{stack_trace}"
return f"""You are a senior software engineer debugging an issue.
Analyze the following error and provide:
1. Root cause analysis
2. Step-by-step fix
3. Prevention recommendations
{context}"""
# Test the prompt locally
result = debug_error("KeyError: 'user_id'", "File app.py, line 42")
print(result)7. Complete Server Example
A minimal but complete MCP server in a single file — ready to save and run.
%%writefile server.py
# server.py — Complete MCP Server
from mcp.server.fastmcp import FastMCP
import httpx
import json
mcp = FastMCP("my-knowledge-server")
@mcp.tool()
async def search_documentation(query: str, max_results: int = 5) -> str:
"""Search the project documentation for relevant articles.
Args:
query: The search query string
max_results: Maximum number of results to return
"""
return f"Search results for: {query}"
@mcp.resource("config://app-settings")
def get_app_settings() -> str:
"""Return the current application configuration."""
return json.dumps({"version": "2.1.0", "environment": "production"}, indent=2)
@mcp.prompt()
def debug_error(error_message: str) -> str:
"""Create a debugging prompt for analyzing errors."""
return f"Analyze this error and suggest a fix:\n{error_message}"
if __name__ == "__main__":
mcp.run(transport="stdio")
print("✅ server.py written. Run with: uv run server.py")8. Real-World Project Knowledge Server
A complete MCP server combining tools, resources, and prompts for project context.
%%writefile project_server.py
# project_server.py — Full project knowledge MCP server
import json
import subprocess
from pathlib import Path
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("project-knowledge")
PROJECT_ROOT = Path("/path/to/your/project")
# ─── Tools ──────────────────────────────────────────
@mcp.tool()
def search_codebase(pattern: str, file_glob: str = "**/*.py") -> str:
"""Search the codebase using grep for a pattern.
Args:
pattern: Regex pattern to search for
file_glob: File glob to limit search scope (default: all Python files)
"""
try:
result = subprocess.run(
["grep", "-rn", "--include", file_glob, pattern, str(PROJECT_ROOT)],
capture_output=True, text=True, timeout=30,
)
if not result.stdout.strip():
return f"No matches found for pattern: {pattern}"
lines = result.stdout.strip().split("\n")[:20]
return "\n".join(lines)
except subprocess.TimeoutExpired:
return "Search timed out. Try a more specific pattern."
@mcp.tool()
def read_file(file_path: str, start_line: int = 1, end_line: int = 100) -> str:
"""Read a file from the project with optional line range.
Args:
file_path: Relative path from project root
start_line: First line to read (1-indexed)
end_line: Last line to read (inclusive)
"""
target = PROJECT_ROOT / file_path
# Security: prevent path traversal
if not target.resolve().is_relative_to(PROJECT_ROOT.resolve()):
return "Error: Access denied — path is outside the project."
if not target.exists():
return f"File not found: {file_path}"
lines = target.read_text().splitlines()
selected = lines[start_line - 1 : end_line]
numbered = [f"{i}: {line}" for i, line in enumerate(selected, start=start_line)]
return "\n".join(numbered)
@mcp.tool()
def run_tests(test_path: str = "", verbose: bool = False) -> str:
"""Run project tests using pytest.
Args:
test_path: Specific test file or directory (default: all tests)
verbose: Whether to show verbose output
"""
cmd = ["python", "-m", "pytest", "--tb=short", "-q"]
if verbose:
cmd.append("-v")
if test_path:
cmd.append(str(PROJECT_ROOT / test_path))
else:
cmd.append(str(PROJECT_ROOT / "tests"))
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, cwd=str(PROJECT_ROOT))
output = result.stdout + result.stderr
if len(output) > 3000:
output = output[:3000] + "\n... (truncated)"
return output
except subprocess.TimeoutExpired:
return "Tests timed out after 120 seconds."
# ─── Resources ──────────────────────────────────────
@mcp.resource("project://readme")
def get_readme() -> str:
"""Return the project README."""
readme = PROJECT_ROOT / "README.md"
return readme.read_text() if readme.exists() else "No README.md found."
@mcp.resource("project://dependencies")
def get_dependencies() -> str:
"""Return the project dependencies."""
for dep_file in ["pyproject.toml", "requirements.txt", "package.json"]:
path = PROJECT_ROOT / dep_file
if path.exists():
return f"# {dep_file}\n{path.read_text()}"
return "No dependency file found."
# ─── Prompts ────────────────────────────────────────
@mcp.prompt()
def code_review(file_path: str) -> str:
"""Generate a prompt for reviewing a specific file."""
return f"""You are a senior engineer conducting a code review.
Review the file at `{file_path}` and provide feedback on:
1. Code quality and readability
2. Potential bugs or edge cases
3. Performance considerations
4. Security concerns
5. Suggestions for improvement"""
@mcp.prompt()
def investigate_bug(description: str) -> str:
"""Generate a prompt for investigating a bug."""
return f"""You are debugging a reported issue. Bug: {description}
1. Use search_codebase to find relevant code
2. Use read_file to examine suspicious areas
3. Use run_tests to verify behavior
4. Provide root cause analysis and fix"""
if __name__ == "__main__":
mcp.run(transport="stdio")
print("✅ project_server.py written")9. Mistral AI Integration
Connect your MCP server to a Mistral agent — their SDK includes built-in MCP client support.
import os
# os.environ["MISTRAL_API_KEY"] = "your-api-key-here" # Uncomment and set
# --- Mistral + MCP Integration ---
# This code connects a Mistral agent to your MCP server
MISTRAL_MCP_CODE = """
import asyncio
import os
from pathlib import Path
from mistralai.client import Mistral
from mistralai.extra.run.context import RunContext
from mcp import StdioServerParameters
from mistralai.extra.mcp.stdio import MCPClientSTDIO
MODEL = "mistral-medium-latest"
cwd = Path(__file__).parent
async def main() -> None:
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
server_params = StdioServerParameters(
command="uv",
args=["--directory", str(cwd), "run", "server.py"],
env=None,
)
agent = client.beta.agents.create(
model=MODEL,
name="project-assistant",
instructions="You help developers understand and work with the project.",
)
async with RunContext(agent_id=agent.id) as run_ctx:
mcp_client = MCPClientSTDIO(stdio_params=server_params)
await run_ctx.register_mcp_client(mcp_client=mcp_client)
result = await client.beta.conversations.run_async(
run_ctx=run_ctx,
inputs="Search for all API endpoint definitions in the project.",
)
for entry in result.output_entries:
print(entry)
if __name__ == "__main__":
asyncio.run(main())
"""
print(MISTRAL_MCP_CODE)10. Docker Deployment
Containerize your MCP server for production deployment.
DOCKERFILE = """
FROM python:3.12-slim
WORKDIR /app
# Install uv
RUN pip install uv
# Copy project files
COPY pyproject.toml uv.lock ./
COPY server.py ./
# Install dependencies
RUN uv sync --frozen
# Expose port for Streamable HTTP transport
EXPOSE 8080
# Run the server
CMD ["uv", "run", "server.py"]
"""
print("Dockerfile for MCP Server:")
print(DOCKERFILE)11. Transport Configuration
MCP supports two transports: stdio (local) and Streamable HTTP (remote).
import os
# Transport-aware server entry point
TRANSPORT_CODE = """
import os
if __name__ == "__main__":
transport = os.environ.get("MCP_TRANSPORT", "stdio")
if transport == "streamable-http":
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=int(os.environ.get("PORT", 8080)),
)
else:
mcp.run(transport="stdio")
"""
print("Transport configuration:")
print(TRANSPORT_CODE)
# --- Client Configuration Examples ---
claude_config = {
"mcpServers": {
"my-knowledge-server": {
"command": "uv",
"args": ["--directory", "/path/to/my-mcp-server", "run", "server.py"]
}
}
}
vscode_config = {
"servers": {
"my-knowledge-server": {
"command": "uv",
"args": ["--directory", "${workspaceFolder}", "run", "server.py"]
}
}
}
print("\n--- Claude Desktop config ---")
print(json.dumps(claude_config, indent=2))
print("\n--- VS Code .vscode/mcp.json ---")
print(json.dumps(vscode_config, indent=2))Security Best Practices
| Concern | Recommendation |
|---|---|
| Path traversal | Validate file paths against an allowed root directory |
| SQL injection | Use parameterized queries; restrict to read-only ops |
| DNS rebinding | Validate the Origin header on HTTP connections |
| Network binding | Bind to 127.0.0.1 for local servers, not 0.0.0.0 |
| Authentication | Use OAuth or bearer tokens for remote servers |
| Rate limiting | Implement per-client rate limits |
MCP vs. Direct Function Calling
| Aspect | Direct Function Calling | MCP |
|---|---|---|
| Protocol | Provider-specific | Open standard (JSON-RPC 2.0) |
| Discovery | Manual schema definition | Automatic via tools/list |
| Portability | Locked to one provider | Works across all MCP hosts |
| Ecosystem | Provider-specific | Growing open ecosystem |