Budowanie pierwszego serwera MCP

Konfiguracja projektu serwera MCP

# Create a project directory
# mkdir my_mcp_server && cd my_mcp_server

# Create a virtual environment
# python -m venv venv && source venv/bin/activate

# Install the MCP SDK
# pip install mcp httpx

# Project structure:
# my_mcp_server/
#   server.py          <- Your MCP server
#   requirements.txt
#   README.md

Tworzenie obiektu serwera

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import httpx

# Create the server — name is shown in Claude Desktop
app = Server('weather-server')

# --- Tool registrations go here ---

# Entry point
if __name__ == '__main__':
    asyncio.run(stdio_server(app))

Rejestrowanie narzędzi za pomocą @app.list_tools()

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name='get_weather',
            description='Get current weather for a city. Use when the user asks about weather in a specific location.',
            inputSchema={
                'type': 'object',
                'properties': {
                    'city': {
                        'type': 'string',
                        'description': 'City name, e.g. London or New York'
                    },
                    'units': {
                        'type': 'string',
                        'enum': ['metric', 'imperial'],
                        'description': 'Temperature units. Default is metric.'
                    }
                },
                'required': ['city']
            }
        ),
        types.Tool(
            name='list_cities',
            description='Return a list of major cities the user can query weather for.',
            inputSchema={'type': 'object', 'properties': {}, 'required': []}
        )
    ]

Implementowanie wykonywania narzędzi

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == 'list_cities':
        cities = ['London', 'New York', 'Tokyo', 'Paris', 'Sydney']
        return [types.TextContent(type='text', text=', '.join(cities))]

    if name == 'get_weather':
        city = arguments['city']
        units = arguments.get('units', 'metric')
        unit_symbol = 'C' if units == 'metric' else 'F'

        # Call real weather API (stub here)
        async with httpx.AsyncClient() as client:
            # Replace with actual API call
            result = f'{city}: 18{chr(176)}{unit_symbol}, partly cloudy, humidity 65%'

        return [types.TextContent(type='text', text=result)]

    raise ValueError(f'Unknown tool: {name}')

Udostępnianie zasobów

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri='weather://supported-cities',
            name='Supported Cities List',
            description='Complete list of cities available in this weather server.',
            mimeType='text/plain'
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == 'weather://supported-cities':
        cities = ['London', 'New York', 'Tokyo', 'Paris', 'Sydney', 'Dubai', 'Singapore']
        return '\n'.join(cities)
    raise ValueError(f'Unknown resource URI: {uri}')

Rejestrowanie szablonów promptów

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name='weather-report',
            description='Generate a formatted weather report for a city.',
            arguments=[
                types.PromptArgument(name='city', description='City name', required=True)
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
    if name == 'weather-report':
        city = arguments.get('city', 'London')
        return types.GetPromptResult(
            description=f'Weather report for {city}',
            messages=[
                types.PromptMessage(
                    role='user',
                    content=types.TextContent(
                        type='text',
                        text=f'Use the get_weather tool to look up {city} and give me a detailed weather report including what clothing I should wear.'
                    )
                )
            ]
        )
    raise ValueError(f'Unknown prompt: {name}')

Łączenie z Claude Desktop

# ~/Library/Application Support/Claude/claude_desktop_config.json
# Add this JSON configuration:

# {
#   "mcpServers": {
#     "weather-server": {
#       "command": "/path/to/venv/bin/python",
#       "args": ["/path/to/my_mcp_server/server.py"],
#       "env": {
#         "WEATHER_API_KEY": "your_api_key_here"
#       }
#     }
#   }
# }

# After saving, restart Claude Desktop.
# Your server's tools will appear in Claude's tool list.

Testowanie serwera za pomocą MCP CLI

# Install the MCP development tools
# pip install 'mcp[cli]'

# Start the inspector with your server
# mcp dev server.py

# The inspector opens at http://localhost:5173
# You can:
# - See all registered tools and their schemas
# - Call tools with custom arguments
# - Browse available resources
# - Test prompt templates
# - View the full JSON-RPC message exchange

Obsługa błędów w serwerach MCP

@app.call_tool()
async def call_tool_safe(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == 'get_weather':
            city = arguments.get('city')
            if not city:
                return [types.TextContent(type='text', text='Error: city argument is required.')]
            result = await fetch_weather(city, arguments.get('units', 'metric'))
            return [types.TextContent(type='text', text=result)]
        raise ValueError(f'Unknown tool: {name}')
    except httpx.TimeoutException:
        return [types.TextContent(type='text', text='Error: Weather API timed out. Try again.')]
    except Exception as e:
        return [types.TextContent(type='text', text=f'Error: {str(e)}')]

Logowanie w serwerach MCP

import logging
import sys

# Configure logging to stderr (NOT stdout — that's the MCP channel)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger('weather-server')

# In your tool handler:
# logger.info(f'Getting weather for {city}')
# logger.error(f'API call failed: {e}')

# Claude Desktop captures stderr to a log file:
# ~/Library/Logs/Claude/mcp-server-weather-server.log

Pakowanie serwera MCP

Szybki test

Podsumowanie lekcji