Building Your First MCP Server: A Developer's Guide

python -m venv mcp-env
source mcp-env/bin/activate  # On Windows: mcp-env\Scripts\activate
# Install MCP SDK with FastMCP

from mcp.server.fastmcp import FastMCP
import math
# Initialize the server with a descriptive name
mcp = FastMCP("Calculator Server")
# Basic arithmetic tools
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b
@mcp.tool()
def calculate_factorial(n: int) -> int:
    """Calculate factorial of a number"""
    if n < 0:
        raise ValueError("Factorial is not defined for negative numbers")
    return math.factorial(n)
# A resource that provides calculation history
@mcp.resource("calculator://history")
def calculation_history() -> str:
    """Get recent calculation history"""
    # In a real implementation, you'd track this
    return "Recent calculations: 5+3=8, 4*2=8, factorial(5)=120"
# A prompt to guide AI behavior
@mcp.prompt()
def calculator_assistant() -> str:
    """Instructions for using the calculator effectively"""
    return """
    You are a helpful math assistant. Use the calculator tools to:
    - Perform accurate calculations for users
    - Show your work step by step
    - Explain mathematical concepts when helpful
    - Check your answers using the available tools
    """
# Run the server
if __name__ == "__main__":
    mcp.run(transport="stdio")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

# For locally developed servers (most common)
npx @modelcontextprotocol/inspector node path/to/your/server.js
# For Python servers
npx @modelcontextprotocol/inspector python path/to/your/server.py
# For servers with specific arguments
npx @modelcontextprotocol/inspector python server.py --config

from mcp.server.fastmcp import FastMCP
import httpx
import json
from typing import Dict, Any
mcp = FastMCP("Weather Service")
async def fetch_weather_data(location: str) -> Dict[str, Any]:
    """Helper function to fetch weather from National Weather Service"""
    # Simplified example - real implementation would handle errors
    async with httpx.AsyncClient() as client:
        # Get coordinates for location (simplified)
        response = await client.get(f"https://api.weather.gov/points/{location}")
        if response.status_code == 200:
            data = response.json()
            forecast_url = data["properties"]["forecast"]
            
            # Get forecast
            forecast_response = await client.get(forecast_url)
            return forecast_response.json()
    
    return {"error": "Could not fetch weather data"}
@mcp.tool()
async def get_forecast(location: str) -> str:
    """Get weather forecast for a specific location (US only)"""
    try:
        weather_data = await fetch_weather_data(location)
        if "error" in weather_data:
            return f"Error: {weather_data['error']}"
        
        # Format the forecast nicely
        periods = weather_data.get("properties", {}).get("periods", [])[:3]
        forecast = []
        
        for period in periods:
            forecast.append(f"{period['name']}: {period['detailedForecast']}")
        
        return "\\n\\n".join(forecast)
    except Exception as e:
        return f"Error getting forecast: {str(e)}"
@mcp.resource("weather://current/{location}")
async def current_conditions(location: str) -> str:
    """Current weather conditions as a resource"""
    data = await fetch_weather_data(location)
    return json.dumps(data, indent=2)
@mcp.prompt()
def weather_assistant_guidance() -> str:
    """Comprehensive weather assistant instructions"""
    return """
    You are a knowledgeable weather assistant. When users ask about weather:
    
    1. Use get_forecast tool for future weather predictions
    2. Access current_conditions resource for detailed current data
    3. Always specify that forecasts are for US locations only
    4. Provide helpful context about weather conditions
    5. Suggest appropriate activities based on the forecast
    6. Alert users to any severe weather conditions
    
    Be conversational and helpful, not just informational.
    """
if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "calculator": {
      "command": "python",
      "args": ["/absolute/path/to/your/calculator.py"],
      "env": {
        "PYTHONPATH": "/path/to/your/project"
      }
    },
    "weather": {
      "command": "/absolute/path/to/venv/bin/python",
      "args": ["/absolute/path/to/weather_server.py"]
    }
  }
}

@mcp.tool()
async def reliable_api_call(query: str) -> str:
    """Example of comprehensive error handling"""
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(f"https://api.example.com/search?q={query}")
            response.raise_for_status()
            
            data = response.json()
            return f"Results: {data.get('results', 'No results found')}"
            
    except httpx.TimeoutException:
        return "Error: Request timed out. Please try again."
    except httpx.HTTPStatusError as e:
        return f"Error: API returned status {e.response.status_code}"
    except json.JSONDecodeError:
        return "Error: Invalid response from API"
    except Exception as e:
        return f"Unexpected error: {str(e)}"

import json
from pathlib import Path
STATE_FILE = Path("server_state.json")
def load_state() -> dict:
    """Load state from disk"""
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}
def save_state(state: dict):
    """Save state to disk"""
    STATE_FILE.write_text(json.dumps(state, indent=2))
@mcp.tool()
def remember_preference(key: str, value: str) -> str:
    """Store a user preference"""
    state = load_state()
    state[key] = value
    save_state(state)
    return f"Remembered: {key} = {value}"

import sqlite3
from contextlib import contextmanager
@contextmanager
def get_db_connection():
    """Context manager for database connections"""
    conn = sqlite3.connect("server.db")
    try:
        yield conn
    finally:
        conn.close()
@mcp.tool()
def store_user_data(user_id: str, data: str) -> str:
    """Store data with proper database handling"""
    try:
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT OR REPLACE INTO user_data (user_id, data) VALUES (?, ?)",
                (user_id, data)
            )
            conn.commit()
            return "Data stored successfully"
    except Exception as e:
        return f"Database error: {str(e)}"

import os
from typing import Optional
def get_api_key(service: str) -> Optional[str]:
    """Safely retrieve API keys from environment"""
    key = os.getenv(f"{service.upper()}_API_KEY")
    if not key:
        raise ValueError(f"Missing {service} API key in environment variables")
    return key
@mcp.tool()
async def secure_api_call(query: str) -> str:
    """Tool that uses API key securely"""
    try:
        api_key = get_api_key("openweather")
        # Use api_key in request...
    except ValueError as e:
        return f"Configuration error: {str(e)}"

from typing import Literal
@mcp.tool()
def validated_calculation(
    operation: Literal["add", "subtract", "multiply", "divide"],
    a: float,
    b: float
) -> str:
    """Calculator with strict input validation"""
    if operation == "divide" and b == 0:
        return "Error: Division by zero"
    
    operations = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b
    }
    
    result = operations[operation]
    return f"{a} {operation} {b} = {result}"

from functools import lru_cache
import time
@lru_cache(maxsize=100)
def expensive_computation(input_data: str) -> str:
    """Cache expensive operations"""
    # Simulate expensive operation
    time.sleep(2)
    return f"Processed: {input_data}"
@mcp.tool()
def cached_tool(data: str) -> str:
    """Tool that benefits from caching"""
    return expensive_computation(data)

import asyncio
from typing import List
@mcp.tool()
async def batch_api_calls(queries: List[str]) -> str:
    """Efficiently handle multiple API calls"""
    async def single_call(query: str) -> str:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"https://api.example.com/search?q={query}")
            return response.json()
    
    # Process all queries concurrently
    results = await asyncio.gather(*[single_call(q) for q in queries])
    return f"Processed {len(results)} queries successfully"

@mcp.tool()
async def create_project_workflow(
    project_name: str, 
    team_members: List[str]
) -> str:
    """Create a complete project setup across multiple services"""
    results = []
    
    # Create Linear project
    linear_project = await create_linear_project(project_name)
    results.append(f"Linear project created: {linear_project['id']}")
    
    # Create Slack channel
    slack_channel = await create_slack_channel(project_name.lower().replace(' ', '-'))
    results.append(f"Slack channel created: #{slack_channel['name']}")
    
    # Add team members to both
    for member in team_members:
        await add_member_to_linear(linear_project['id'], member)
        await invite_to_slack_channel(slack_channel['id'], member)
    
    # Create initial Notion page
    notion_page = await create_notion_project_page(project_name, linear_project['id'])
    results.append(f"Notion documentation created: {notion_page['url']}")
    
    return "\\n".join(results)

def create_api_tools(api_config: dict):
    """Generate MCP tools from API configuration"""
    for endpoint in api_config['endpoints']:
        tool_name = f"call_{endpoint['name']}"
        
        @mcp.tool(name=tool_name)
        async def dynamic_api_call(
            **kwargs
        ) -> str:
            """Dynamically generated API tool"""
            async with httpx.AsyncClient() as client:
                response = await client.request(
                    method=endpoint['method'],
                    url=f"{api_config['base_url']}{endpoint['path']}",
                    **kwargs
                )
                return response.json()
        
        # Register the tool
        globals()[tool_name] = dynamic_api_call
# Load configuration and generate tools
api_config = load_api_configuration()
create_api_tools(api_config)

from datetime import datetime
import uuid
class ConversationMemory:
    def __init__(self):
        self.conversations = {}
    
    def get_or_create_session(self, user_id: str) -> str:
        """Get existing session or create new one"""
        if user_id not in self.conversations:
            session_id = str(uuid.uuid4())
            self.conversations[user_id] = {
                'session_id': session_id,
                'created_at': datetime.now(),
                'messages': [],
                'context': {}
            }
        return self.conversations[user_id]['session_id']
memory = ConversationMemory()
@mcp.tool()
def remember_context(user_id: str, key: str, value: str) -> str:
    """Store context for the user's session"""
    session_id = memory.get_or_create_session(user_id)
    memory.conversations[user_id]['context'][key] = value
    return f"Remembered {key} for session {session_id}"
@mcp.resource("memory://context/{user_id}")
def get_user_context(user_id: str) -> str:
    """Retrieve user's conversation context"""
    if user_id in memory.conversations:
        context = memory.conversations[user_id]['context']
        return json.dumps(context, indent=2)
    return "{}"

import consul
@mcp.tool()
async def discover_services(service_type: str) -> List[str]:
    """Discover available services in the infrastructure"""
    c = consul.Consul()
    services = c.health.service(service_type, passing=True)[1]
    return [f"{s['Service']['Address']}:{s['Service']['Port']}" 
            for s in services]

@mcp.tool()
async def execute_deployment_workflow(
    app_name: str, 
    environment: str
) -> str:
    """Orchestrate complex deployment workflow"""
    workflow_steps = [
        ("build", build_application),
        ("test", run_tests),
        ("deploy", deploy_to_environment),
        ("verify", verify_deployment),
        ("notify", send_notifications)
    ]
    
    results = []
    for step_name, step_func in workflow_steps:
        try:
            result = await step_func(app_name, environment)
            results.append(f"✅ {step_name}: {result}")
        except Exception as e:
            results.append(f"❌ {step_name}: {str(e)}")
            # Handle rollback logic
            await rollback_deployment(app_name, environment)
            break
    
    return "\\n".join(results)

from mcp.server.fastmcp import FastMCP
import os
import httpx
import json
# Initialize server
mcp = FastMCP("My Custom Server")
# Load configuration
API_KEY = os.getenv("MY_API_KEY")
if not API_KEY:
    raise ValueError("MY_API_KEY environment variable required")
@mcp.tool()
async def my_first_tool(input_data: str) -> str:
    """Description of what this tool does"""
    try:
        # Your logic here
        return f"Processed: {input_data}"
    except Exception as e:
        return f"Error: {str(e)}"
@mcp.resource("data://my-resource")
async def my_resource() -> str:
    """Description of this resource"""
    return json.dumps({"status": "active", "data": "sample"})
@mcp.prompt()
def usage_guidance() -> str:
    """How to use this server effectively"""
    return """
    This server provides tools for [your use case].
    Use my_first_tool to [specific functionality].
    Check my_resource for [what it provides].
    """
if __name__ == "__main__":
    mcp.run(transport="stdio")