Skip to main content

**Hands-On MCP Server — Your First Tool

Calculating read time…

MCP Server = Tool Provider

Think of it as a store that sells tools.

It has tools on shelves, waiting for customers.

MCP Client = Tool User

This is the AI that comes shopping.

It asks: "What tools do you have?" then uses them.

💡 Quick Analogy: Restaurant vs Customer

Server = Restaurant with menu (available tools)

Client = Customer who orders from menu

Today, you're opening your first restaurant!

Step 1: Setting Up Your Python Environment

First, let's create a clean workspace.

Create Your Project Folder

# Open terminal/command prompt
mkdir my-first-mcp-server
cd my-first-mcp-server

# Create virtual environment
python -m venv venv

# Activate it (choose your OS):
# Windows:
venv\Scripts\activate

# Mac/Linux:
source venv/bin/activate

✅ DO use virtual environments

They keep your projects isolated.

No conflicts between different Python packages.

Like having separate toolboxes for different jobs.

Install Required Packages

We need the MCP SDK:

pip install mcp

Create a requirements.txt file for future reference:

echo "mcp>=0.1.0" > requirements.txt

Step 2: Understanding MCP Server Structure

Every MCP server has three essential parts:

1. SERVER SETUP
   ├── Import MCP libraries
   ├── Create server instance
   └── Configure communication

2. TOOL REGISTRATION
   ├── Define what tools are available
   ├── Describe each tool's purpose
   └── Specify required inputs

3. TOOL IMPLEMENTATION
   ├── Write the actual code
   ├── Handle inputs properly
   └── Return results in MCP format

Step 3: Your First MCP Server Code

Create a file called calculator_server.py:

# calculator_server.py
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import asyncio

📚 Import Breakdown:

  • Server: Main server class
  • InitializationOptions: Server configuration
  • stdio: Communication method (pipes)
  • asyncio: Async programming (MCP uses this)

Creating the Server Instance

Add this to your file:

# Create server instance
server = Server("calculator-server")

# Define our first tool
@server.list_tools()
async def list_tools():
    return [
        {
            "name": "add_numbers",
            "description": "Add two numbers together",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "First number"
                    },
                    "b": {
                        "type": "number", 
                        "description": "Second number"
                    }
                },
                "required": ["a", "b"]
            }
        }
    ]

✅ DO understand the decorator:

@server.list_tools() tells MCP:

"This function returns our tool menu."

Every MCP server needs this function.

What's Happening Here?

Let's break down the tool definition:

{
    "name": "add_numbers",           # Tool's ID (how AI calls it)
    "description": "Add two...",      # What it does (AI reads this)
    "inputSchema": {                  # What inputs it needs
        "type": "object",
        "properties": {               # List of inputs
            "a": {
                "type": "number",     # Must be a number
                "description": "First number"  # Help text
            },
            "b": { ... }
        },
        "required": ["a", "b"]        # Both inputs are required
    }
}

Step 4: Implementing the Actual Tool

Now we write the code that does the actual work.

Add this to your file:

# Implement the add_numbers tool
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "add_numbers":
        # Extract inputs
        a = arguments.get("a")
        b = arguments.get("b")
        
        # Do the calculation
        result = a + b
        
        # Return in MCP format
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }
    
    # If tool not found
    raise ValueError(f"Unknown tool: {name}")

Understanding the Tool Logic

The function flow:

1. AI asks for "add_numbers" with a=5, b=3
2. Function receives name="add_numbers", arguments={"a":5, "b":3}
3. Checks if name matches our tool
4. Extracts a and b from arguments
5. Calculates 5 + 3 = 8
6. Returns {"content": [{"type": "text", "text": "8"}]}

🔍 Important Detail: The return format

MCP expects content array with items.

Each item has type and actual content.

"type": "text" means we're returning plain text.

Step 5: Adding the Server Runner

We need code to start and run the server.

Add this to the bottom of your file:

async def main():
    # Run server with stdio transport
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="calculator-server",
                server_version="1.0.0"
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

Step 6: Complete Server Code

Here's everything together:

# calculator_server.py - COMPLETE CODE
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import asyncio

# Create server instance
server = Server("calculator-server")

# List available tools
@server.list_tools()
async def list_tools():
    return [
        {
            "name": "add_numbers",
            "description": "Add two numbers together",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "First number"
                    },
                    "b": {
                        "type": "number", 
                        "description": "Second number"
                    }
                },
                "required": ["a", "b"]
            }
        }
    ]

# Handle tool calls
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "add_numbers":
        a = arguments.get("a")
        b = arguments.get("b")
        result = a + b
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }
    
    raise ValueError(f"Unknown tool: {name}")

# Server runner
async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="calculator-server",
                server_version="1.0.0"
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

✅ DO save this file exactly as shown

Make sure indentation is correct.

Python is picky about spaces vs tabs.

Copy-paste to avoid typos.

Step 7: Testing Your Server

You don't need an AI to test! Let's do manual testing.

Method 1: Run and Check

First, run your server:

python calculator_server.py

It will start and wait for connections.

This is normal! It's listening for clients.

⚠️ If you get errors:

  • Check Python version (needs 3.8+)
  • Verify MCP installed correctly
  • Ensure virtual environment is active
  • Check for typos in code

Method 2: Create a Test Client

Create a new file test_client.py:

# test_client.py
import subprocess
import json
import time

def send_mcp_message(message):
    """Send a message to our server"""
    # Start server process
    proc = subprocess.Popen(
        ['python', 'calculator_server.py'],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    
    # Send message
    proc.stdin.write(json.dumps(message) + '\n')
    proc.stdin.flush()
    
    # Read response
    response = proc.stdout.readline()
    
    # Clean up
    proc.terminate()
    
    return json.loads(response)

# Test initialization
init_message = {
    "jsonrpc": "2.0",
    "method": "initialize",
    "params": {
        "clientInfo": {
            "name": "test-client",
            "version": "1.0.0"
        }
    },
    "id": 1
}

print("Testing initialization...")
response = send_mcp_message(init_message)
print("Response:", json.dumps(response, indent=2))

Step 8: Understanding What Just Happened

Let's visualize the server's lifecycle:

Server Lifecycle

┌─────────────────────────────────────────────┐
│            SERVER STARTS                    │
│ 1. Import libraries, create Server instance │
│ 2. Define @server.list_tools() function     │
│ 3. Define @server.call_tool() function      │
│ 4. Wait for client connections (stdio)      │
└─────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────┐
│        CLIENT CONNECTS                      │
│ 1. Client sends "initialize" request        │
│ 2. Server responds with capabilities        │
│ 3. Client asks "what tools?" (tools/list)   │
│ 4. Server returns tool definitions          │
└─────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────┐
│          TOOL EXECUTION                     │
│ 1. Client calls "add_numbers" with a,b      │
│ 2. Server's call_tool() function runs       │
│ 3. Calculation happens (a + b)              │
│ 4. Result formatted and returned            │
└─────────────────────────────────────────────┘

Step 9: Testing the Add Tool

Let's create a proper test.

Update your test_client.py:

# Extended test
def test_add_tool():
    print("\n=== Testing add_numbers tool ===")
    
    # First, get available tools
    list_message = {
        "jsonrpc": "2.0",
        "method": "tools/list",
        "params": {},
        "id": 2
    }
    
    print("1. Listing tools...")
    tools_response = send_mcp_message(list_message)
    print("Available tools:", json.dumps(tools_response, indent=2))
    
    # Now call add_numbers
    add_message = {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {
            "name": "add_numbers",
            "arguments": {
                "a": 42,
                "b": 17
            }
        },
        "id": 3
    }
    
    print("\n2. Calling add_numbers(42, 17)...")
    result = send_mcp_message(add_message)
    print("Result:", json.dumps(result, indent=2))
    
    # Extract the actual answer
    content = result.get("result", {}).get("content", [])
    if content:
        answer = content[0].get("text", "No answer found")
        print(f"\n✅ 42 + 17 = {answer}")
    else:
        print("\n❌ Failed to get answer")

if __name__ == "__main__":
    test_add_tool()

❌ DON'T skip testing

Testing ensures your server works correctly.

AI will fail mysteriously if your server has bugs.

Always test manually first.

Step 10: Your Exercise - Add Subtraction

🏆 Challenge: Add subtract_numbers Tool

Your mission: Extend the server with subtraction.

Requirements:

  1. Add "subtract_numbers" to the tools list
  2. Implement the subtraction logic in call_tool()
  3. Test it works correctly (50 - 8 = 42)

Hints:

  • Copy the add_numbers pattern
  • Change the name and description
  • Subtraction formula: a - b
  • Add new if condition in call_tool()

Success criteria:

  • Server starts without errors
  • Both add and subtract appear in tools list
  • subtract_numbers(50, 8) returns 42

Step-by-Step Guidance

Here's how to approach the exercise:

# Step 1: Update list_tools() function
# Add this to the tools list (alongside add_numbers):
{
    "name": "subtract_numbers",
    "description": "Subtract second number from first number",
    "inputSchema": {
        "type": "object",
        "properties": {
            "a": {
                "type": "number",
                "description": "Number to subtract from"
            },
            "b": {
                "type": "number",
                "description": "Number to subtract"
            }
        },
        "required": ["a", "b"]
    }
}

# Step 2: Update call_tool() function
# Add this condition (after the add_numbers check):
if name == "subtract_numbers":
    a = arguments.get("a")
    b = arguments.get("b")
    result = a - b  # This is the key change!
    
    return {
        "content": [
            {
                "type": "text",
                "text": str(result)
            }
        ]
    }

Common Issues & Solutions

Issue 1: Server Doesn't Start

🔧 Troubleshooting:

  • Error: "No module named 'mcp'"
  • Solution: Activate virtual env, run pip install mcp
  • Error: SyntaxError
  • Solution: Check Python version ≥ 3.8
  • Server starts but immediately exits
  • Solution: Ensure asyncio.run(main()) is at bottom

Issue 2: Tools Not Listed

If AI can't see your tools:

  1. Check @server.list_tools() decorator exists
  2. Verify function returns a list (not a dictionary)
  3. Ensure tool names match exactly (case-sensitive)
  4. Check JSON structure is valid

Issue 3: Tool Doesn't Work

If tool is listed but fails when called:

  1. Check @server.call_tool() decorator
  2. Verify name == "your_tool_name" matches exactly
  3. Ensure you're extracting arguments correctly
  4. Check return format matches MCP spec

Advanced Concepts: Leveling Up

Adding Multiple Tools Efficiently

Instead of giant if-else chains, use a dictionary:

# Define tool implementations
tool_functions = {
    "add_numbers": lambda a, b: a + b,
    "subtract_numbers": lambda a, b: a - b,
    "multiply_numbers": lambda a, b: a * b,
    "divide_numbers": lambda a, b: a / b if b != 0 else "Error: division by zero"
}

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name in tool_functions:
        a = arguments.get("a")
        b = arguments.get("b")
        result = tool_functions[name](a, b)
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }
    
    raise ValueError(f"Unknown tool: {name}")

Adding Input Validation

Make your server robust:

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "divide_numbers":
        a = arguments.get("a")
        b = arguments.get("b")
        
        # Validate inputs
        if b == 0:
            return {
                "content": [
                    {
                        "type": "text",
                        "text": "Error: Cannot divide by zero"
                    }
                ],
                "isError": True
            }
        
        result = a / b
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }

Testing Without AI: Advanced Techniques

Create a Comprehensive Test Suite

# advanced_test.py
import json
import subprocess
import time

class MCPTester:
    def __init__(self, server_script):
        self.server_script = server_script
        self.process = None
        
    def start_server(self):
        """Start the MCP server"""
        self.process = subprocess.Popen(
            ['python', self.server_script],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1
        )
        time.sleep(0.5)  # Give server time to start
        
    def send_request(self, request):
        """Send a request and get response"""
        request_str = json.dumps(request) + '\n'
        self.process.stdin.write(request_str)
        self.process.stdin.flush()
        
        # Read response
        response_line = self.process.stdout.readline()
        return json.loads(response_line.strip())
    
    def stop_server(self):
        """Stop the server"""
        if self.process:
            self.process.terminate()
            self.process.wait()
    
    def run_full_test(self):
        """Run complete test sequence"""
        print("🚀 Starting MCP Server Tests...")
        
        self.start_server()
        
        try:
            # Test 1: Initialize
            print("\n1. Testing initialization...")
            init_response = self.send_request({
                "jsonrpc": "2.0",
                "method": "initialize",
                "params": {
                    "clientInfo": {"name": "tester", "version": "1.0.0"}
                },
                "id": 1
            })
            print(f"   ✓ Server: {init_response.get('result', {}).get('serverInfo', {}).get('name')}")
            
            # Test 2: List tools
            print("\n2. Testing tool listing...")
            tools_response = self.send_request({
                "jsonrpc": "2.0",
                "method": "tools/list",
                "params": {},
                "id": 2
            })
            tools = tools_response.get('result', {}).get('tools', [])
            print(f"   ✓ Found {len(tools)} tools:")
            for tool in tools:
                print(f"     - {tool['name']}: {tool['description']}")
            
            # Test 3: Test each tool
            print("\n3. Testing tool execution...")
            test_cases = [
                ("add_numbers", {"a": 10, "b": 5}, "15"),
                ("subtract_numbers", {"a": 10, "b": 5}, "5"),
                ("multiply_numbers", {"a": 10, "b": 5}, "50"),
                ("divide_numbers", {"a": 10, "b": 5}, "2.0"),
            ]
            
            for tool_name, args, expected in test_cases:
                if any(t['name'] == tool_name for t in tools):
                    result = self.send_request({
                        "jsonrpc": "2.0",
                        "method": "tools/call",
                        "params": {
                            "name": tool_name,
                            "arguments": args
                        },
                        "id": 3
                    })
                    actual = result.get('result', {}).get('content', [{}])[0].get('text', '')
                    status = "✓" if str(actual) == expected else "✗"
                    print(f"   {status} {tool_name}({args['a']}, {args['b']}) = {actual} (expected: {expected})")
            
            print("\n🎉 All tests completed!")
            
        finally:
            self.stop_server()

# Run tests
if __name__ == "__main__":
    tester = MCPTester("calculator_server.py")
    tester.run_full_test()

Production-Ready Improvements

Add Logging

import logging

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("calculator-server")

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info(f"Tool called: {name} with args: {arguments}")
    
    if name == "add_numbers":
        result = arguments.get("a") + arguments.get("b")
        logger.info(f"Calculation: {arguments['a']} + {arguments['b']} = {result}")
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": str(result)
                }
            ]
        }

Add Error Handling

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "add_numbers":
            # Validate inputs exist
            if "a" not in arguments or "b" not in arguments:
                raise ValueError("Missing required parameters: a and b")
            
            # Validate types
            if not isinstance(arguments["a"], (int, float)):
                raise TypeError("Parameter 'a' must be a number")
            if not isinstance(arguments["b"], (int, float)):
                raise TypeError("Parameter 'b' must be a number")
            
            result = arguments["a"] + arguments["b"]
            
            return {
                "content": [
                    {
                        "type": "text",
                        "text": str(result)
                    }
                ]
            }
    
    except Exception as e:
        # Return error in MCP format
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"Error: {str(e)}"
                }
            ],
            "isError": True
        }
Labels:

Comments

Post a Comment