MCP SDK

@modelcontextprotocol/server
├── McpServer          - High-level API (рекомендуется)
├── Server             - Low-level implementation
├── StdioServerTransport
├── StreamableHTTPServerTransport
└── SSEServerTransport (deprecated)

import { McpServer } from '@modelcontextprotocol/server';
import { StdioServerTransport } from '@modelcontextprotocol/server/stdio';
import * as z from 'zod';

// Создание сервера
const server = new McpServer({
  name: 'my-mcp-server',
  version: '1.0.0'
});

// Регистрация инструмента
server.registerTool(
  'greet',
  {
    title: 'Greeting Tool',
    description: 'Returns a greeting message',
    inputSchema: {
      name: z.string().describe('Name to greet')
    }
  },
  async ({ name }) => ({
    content: [{ type: 'text', text: `Hello, ${name}!` }]
  })
);

// Регистрация ресурса
server.registerResource(
  'config',
  'config://app',
  {
    title: 'App Configuration',
    description: 'Application settings',
    mimeType: 'application/json'
  },
  async (uri) => ({
    contents: [{
      uri: uri.href,
      text: JSON.stringify({ version: '1.0', debug: false })
    }]
  })
);

// Регистрация prompt
server.registerPrompt(
  'code-review',
  {
    title: 'Code Review',
    description: 'Review code for best practices',
    argsSchema: {
      code: z.string().describe('Code to review'),
      language: z.string().optional()
    }
  },
  ({ code, language }) => ({
    messages: [{
      role: 'user',
      content: {
        type: 'text',
        text: `Review this ${language || ''} code:\n\n${code}`
      }
    }]
  })
);

// Запуск с stdio транспортом
const transport = new StdioServerTransport();
await server.connect(transport);

import express from 'express';
import { randomUUID } from 'node:crypto';
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/server';

const app = express();
app.use(express.json());

// Stateless режим (рекомендуется для production)
app.post('/mcp', async (req, res) => {
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: () => randomUUID()
  });
  await server.connect(transport);
  await transport.handleRequest(req, res, req.body);
});

app.listen(3000, () => {
  console.log('MCP server running on http://localhost:3000/mcp');
});

server.registerTool(
  'get_weather',
  {
    title: 'Weather Tool',
    description: 'Get weather data',
    inputSchema: {
      city: z.string()
    },
    outputSchema: {
      temperature: z.number(),
      condition: z.string(),
      humidity: z.number()
    }
  },
  async ({ city }) => {
    const data = await fetchWeather(city);
    return {
      content: [{ type: 'text', text: JSON.stringify(data) }],
      structuredContent: data  // Типизированный вывод
    };
  }
);

from mcp.server.fastmcp import FastMCP

# Создание сервера
mcp = FastMCP("My MCP Server", json_response=True)

# Регистрация инструмента
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

# Регистрация ресурса с шаблоном URI
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting."""
    return f"Hello, {name}!"

# Регистрация prompt
@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """Generate a code review prompt."""
    return f"Please review this {language} code:\n\n{code}"

# Запуск
if __name__ == "__main__":
    mcp.run(transport="stdio")  # или "streamable-http"

from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather Service")

class WeatherData(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    humidity: float = Field(description="Humidity percentage")
    condition: str = Field(description="Weather condition")

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """Get current weather for a city."""
    # Автоматическая генерация outputSchema из Pydantic модели
    return WeatherData(
        temperature=22.5,
        humidity=65.0,
        condition="Partly cloudy"
    )

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

mcp = FastMCP("Progress Example")

@mcp.tool()
async def long_task(
    task_name: str,
    ctx: Context[ServerSession, None],
    steps: int = 5
) -> str:
    """Execute a task with progress reporting."""
    await ctx.info(f"Starting: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"Step {i + 1}/{steps}"
        )

    return f"Task '{task_name}' completed"

from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppContext:
    db: Database

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Initialize resources on startup, cleanup on shutdown."""
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()

mcp = FastMCP("Database App", lifespan=app_lifespan)

@mcp.tool()
def query_users(ctx: Context[ServerSession, AppContext]) -> list:
    """Query users from database."""
    db = ctx.request_context.lifespan_context.db
    return db.query("SELECT * FROM users")

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using ModelContextProtocol.Server;
using System.ComponentModel;

var builder = Host.CreateApplicationBuilder(args);

// Добавление MCP сервера
builder.Services
    .AddMcpServer()
    .WithStdioServerTransport()
    .WithToolsFromAssembly();

await builder.Build().RunAsync();

// Определение инструмента через атрибуты
[McpServerToolType]
public static class MathTools
{
    [McpServerTool]
    [Description("Add two numbers together")]
    public static int Add(
        [Description("First number")] int a,
        [Description("Second number")] int b) => a + b;

    [McpServerTool]
    [Description("Multiply two numbers")]
    public static int Multiply(int a, int b) => a * b;
}

using ModelContextProtocol.Server;
using System.ComponentModel;

var builder = WebApplication.CreateBuilder(args);

// Добавление MCP с HTTP транспортом
builder.Services.AddMcpServer()
    .WithHttpTransport()
    .WithToolsFromAssembly();

var app = builder.Build();

// Регистрация MCP endpoint
app.MapMcp();

app.Run("http://localhost:3001");

[McpServerToolType]
public class WeatherTools
{
    [McpServerTool]
    [Description("Get weather forecast")]
    public static async Task<string> GetForecast(
        HttpClient client,  // Инъекция через DI
        [Description("City name")] string city)
    {
        var response = await client.GetStringAsync(
            $"https://api.weather.com/forecast?city={city}"
        );
        return response;
    }
}

[McpServerToolType]
public class DatabaseTools
{
    private readonly ILogger<DatabaseTools> _logger;
    private readonly IDbConnection _db;

    public DatabaseTools(ILogger<DatabaseTools> logger, IDbConnection db)
    {
        _logger = logger;
        _db = db;
    }

    [McpServerTool]
    [Description("Execute SQL query")]
    public async Task<string> Query(string sql)
    {
        _logger.LogInformation("Executing: {Sql}", sql);
        var result = await _db.QueryAsync(sql);
        return JsonSerializer.Serialize(result);
    }
}

[McpServerPromptType]
public static class CodePrompts
{
    [McpServerPrompt]
    [Description("Generate a code review prompt")]
    public static ChatMessage ReviewCode(
        [Description("Code to review")] string code) =>
        new(ChatRole.User, $"Please review this code:\n\n{code}");
}

server.registerTool('risky_operation', schema, async (args) => {
  try {
    const result = await performOperation(args);
    return {
      content: [{ type: 'text', text: JSON.stringify(result) }]
    };
  } catch (error) {
    return {
      content: [{
        type: 'text',
        text: `Error: ${error.message}`
      }],
      isError: true
    };
  }
});

@mcp.tool()
def process_data(data: str, max_items: int = 100) -> str:
    """Process data with validation."""
    if max_items < 1 or max_items > 1000:
        raise ValueError("max_items must be between 1 and 1000")

    # Обработка...
    return "Processed"