MCP сървър с Python: Пълно ръководство за свързване на AI агенти с външни инструменти

Въведение: Защо MCP променя правилата на играта за AI агентите

Ако разработвате AI приложения през 2026 г., почти сигурно вече сте се сблъсквали с този проблем: имате си чудесен LLM, който разсъждава блестящо, генерира код и анализира текст, но в момента, в който трябва да се свърже с база данни, API или файлова система — нещата стават... неудобни. Доскоро единственото решение беше ръчно писане на интеграции за всеки инструмент поотделно. Резултатът? Кошмар от N×M връзки — по една специална интеграция за всяка комбинация от модел и инструмент.

Честно казано, това беше изморително.

Model Context Protocol (MCP) решава точно този проблем. Създаден от Anthropic през ноември 2024 г. и вече предаден на Linux Foundation (чрез Agentic AI Foundation, с подкрепата на OpenAI, Google и Microsoft), MCP е отворен стандарт за комуникация между AI модели и външни системи. Аналогията, която всички обичат да ползват, е USB-C за AI — универсален конектор, който позволява на всяко AI приложение да говори с всяка външна система чрез единен протокол.

Числата са впечатляващи: към март 2026 г. FastMCP (основната Python библиотека за MCP) се изтегля над милион пъти на ден и захранва около 70% от MCP сървърите на всички езици. OpenAI, Google DeepMind, Microsoft и Salesforce вече ползват протокола в продуктите си.

В тази статия ще изградим MCP сървър от нулата с Python. Ще разберем как работи архитектурата, ще създадем реални инструменти за бази данни и API-та, и ще стигнем до продукционно разгръщане с автентикация и сигурност. Хайде да започнем.

Какво представлява Model Context Protocol (MCP)

MCP дефинира клиент-сървър архитектура с три основни компонента:

• MCP Host — AI приложението (например Claude Desktop, Cursor, VS Code), което координира един или повече MCP клиенти
• MCP Client — компонент, поддържащ връзка с MCP сървър и предаващ контекст на хоста
• MCP Server — програма, която предоставя контекст (инструменти, ресурси, шаблони) на клиентите

Протоколът използва JSON-RPC 2.0 за обмен на съобщения. Ако ви звучи познато — да, вдъхновен е от Language Server Protocol (LSP), който реши подобен N×M проблем за код редактори и програмни езици.

Основните примитиви на MCP

Всеки MCP сървър може да предоставя три типа възможности:

• Tools (Инструменти) — изпълними функции, които LLM може да извика (нещо като POST заявки). Примери: изпълнение на SQL заявка, изпращане на имейл, създаване на файл
• Resources (Ресурси) — данни за четене, които LLM зарежда в контекста си (нещо като GET заявки). Примери: съдържание на файл, записи от база данни, конфигурации
• Prompts (Шаблони) — предварително дефинирани шаблони за взаимодействие с LLM, улесняващи потребителите при конкретни задачи

Ключовата разлика между MCP сървър и обикновен REST API е в предназначението. MCP сървърът е проектиран специално за AI модели — предоставя метаданни и структурирани описания, които помагат на модела да разбере кои инструменти са налични, какво правят и кога да ги използва. Моделът сам решава какво да направи, вместо да разчита на твърдо зададени API извиквания. Това е голямата разлика.

Подготовка на средата за разработка

За изграждане на MCP сървър с Python ще ви трябват:

• Python 3.10 или по-нова версия
• uv (препоръчителен пакетен мениджър за Python) или pip
• MCP клиент за тестване — Claude Desktop, Cursor или VS Code с Copilot

Нека инициализираме проекта:

# Инициализиране на проекта с uv
uv init mcp-demo-server
cd mcp-demo-server

# Добавяне на MCP SDK и FastMCP
uv add "mcp[cli]"

# Или с pip:
# pip install "mcp[cli]"

Можете да използвате и самостоятелния пакет FastMCP (версия 3.1.1 към март 2026 г.), който предлага допълнителни възможности извън официалния SDK:

# Самостоятелен FastMCP
uv add fastmcp

# Или с pip:
# pip install fastmcp

Първият ви MCP сървър: Стъпка по стъпка

Достатъчно теория. Нека създадем MCP сървър, който предоставя инструменти за работа с бази данни. Започваме с минимален пример и после ще го разширим.

Минимален MCP сървър

from mcp.server.fastmcp import FastMCP

# Създаване на MCP сървър
mcp = FastMCP("DemoServer", json_response=True)

# Регистриране на инструмент
@mcp.tool()
def add(a: int, b: int) -> int:
    """Събиране на две числа"""
    return a + b

# Регистриране на динамичен ресурс
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Персонализиран поздрав"""
    return f"Здравейте, {name}! Добре дошли в MCP сървъра."

# Регистриране на шаблон
@mcp.prompt()
def summarize(text: str) -> str:
    """Шаблон за обобщаване на текст"""
    return f"Моля, обобщи следния текст в 3 изречения:\n\n{text}"

# Стартиране със stdio транспорт (за локално тестване)
if __name__ == "__main__":
    mcp.run(transport="stdio")

Забележете колко просто е — декорирате обикновени Python функции с @mcp.tool(), @mcp.resource() или @mcp.prompt(), и FastMCP автоматично генерира JSON схемата, валидацията и документацията. Типовите анотации на Python директно се превръщат в описания на параметрите за LLM. Никакъв допълнителен boilerplate.

Свързване с Claude Desktop

За да тествате сървъра, конфигурирайте го в Claude Desktop. Отворете файла claude_desktop_config.json и добавете:

{
  "mcpServers": {
    "demo-server": {
      "command": "uv",
      "args": [
        "--directory",
        "/абсолютен/път/до/mcp-demo-server",
        "run",
        "server.py"
      ]
    }
  }
}

След рестартиране на Claude Desktop, AI асистентът автоматично ще открие наличните инструменти и ще може да ги използва при нужда. Лично на мен ми отнемат буквално секунди — а удовлетворението от „AI-то ми вижда инструментите" е трудно за описване.

Реален пример: MCP сървър за PostgreSQL база данни

Минималният пример е добър за старт, но нека изградим нещо по-сериозно — MCP сървър, който дава на AI агент достъп до PostgreSQL база данни за четене и анализ.

import asyncpg
from mcp.server.fastmcp import FastMCP
from contextlib import asynccontextmanager
from typing import Any

# Конфигурация на сървъра
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

mcp = FastMCP("PostgreSQL Explorer", json_response=True)

# Управление на връзката с базата данни чрез lifespan
@asynccontextmanager
async def get_db_pool():
    pool = await asyncpg.create_pool(DATABASE_URL, min_size=2, max_size=10)
    try:
        yield pool
    finally:
        await pool.close()

# Инструмент за изпълнение на SELECT заявки (само за четене)
@mcp.tool()
async def query_database(sql: str) -> list[dict[str, Any]]:
    """Изпълнява SELECT заявка към PostgreSQL базата данни.
    ВАЖНО: Само SELECT заявки са разрешени за безопасност.
    Примери: SELECT * FROM users LIMIT 10
    """
    # Валидация — само SELECT заявки
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return [{"error": "Само SELECT заявки са разрешени."}]

    # Блокиране на опасни ключови думи
    dangerous_keywords = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "TRUNCATE"]
    for keyword in dangerous_keywords:
        if keyword in normalized:
            return [{"error": f"Заявката съдържа забранена операция: {keyword}"}]

    async with get_db_pool() as pool:
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            return [dict(row) for row in rows]

# Ресурс за списък с таблиците
@mcp.resource("database://tables")
async def list_tables() -> str:
    """Списък на всички таблици в базата данни"""
    async with get_db_pool() as pool:
        async with pool.acquire() as conn:
            rows = await conn.fetch("""
                SELECT table_name, table_type
                FROM information_schema.tables
                WHERE table_schema = 'public'
                ORDER BY table_name
            """)
            result = "Таблици в базата данни:\n"
            for row in rows:
                result += f"- {row['table_name']} ({row['table_type']})\n"
            return result

# Ресурс за схемата на конкретна таблица
@mcp.resource("database://schema/{table_name}")
async def get_table_schema(table_name: str) -> str:
    """Описание на колоните за дадена таблица"""
    # Защита от SQL инжекция — валидиране на името
    if not table_name.isidentifier():
        return "Невалидно име на таблица."

    async with get_db_pool() as pool:
        async with pool.acquire() as conn:
            rows = await conn.fetch("""
                SELECT column_name, data_type, is_nullable, column_default
                FROM information_schema.columns
                WHERE table_name = $1 AND table_schema = 'public'
                ORDER BY ordinal_position
            """, table_name)
            result = f"Схема на таблица '{table_name}':\n"
            for row in rows:
                nullable = "NULL" if row["is_nullable"] == "YES" else "NOT NULL"
                result += f"- {row['column_name']}: {row['data_type']} {nullable}\n"
            return result

# Шаблон за анализ на данни
@mcp.prompt()
def analyze_table(table_name: str) -> str:
    """Шаблон за пълен анализ на таблица от базата данни"""
    return f"""Анализирай таблица '{table_name}' от базата данни. Следвай тези стъпки:
1. Първо провери схемата на таблицата чрез ресурса database://schema/{table_name}
2. Изпълни SELECT COUNT(*) за да видиш броя на записите
3. Покажи примерни записи с LIMIT 5
4. Обобщи структурата и характеристиките на данните"""

if __name__ == "__main__":
    mcp.run(transport="stdio")

Този сървър дава на AI агента способност да разглежда структурата на базата данни, да изпълнява заявки за четене и да анализира данни — всичко с вградена защита срещу деструктивни операции и SQL инжекции. Доста солидна основа, нали?

Реален пример: MCP сървър за REST API интеграция

Другият супер честест сценарий е свързване на AI агент към REST API — вътрешен или външен. Ето пример с интеграция към GitHub API:

import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any
import os

mcp = FastMCP("GitHub Integration", json_response=True)

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
BASE_URL = "https://api.github.com"

def github_headers() -> dict[str, str]:
    return {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json",
        "User-Agent": "MCP-GitHub-Server"
    }

@mcp.tool()
async def list_repo_issues(
    owner: str,
    repo: str,
    state: str = "open",
    limit: int = 10
) -> list[dict[str, Any]]:
    """Списък на issues от GitHub хранилище.
    Параметри:
    - owner: собственик на хранилището (напр. 'anthropics')
    - repo: име на хранилището (напр. 'claude-code')
    - state: 'open', 'closed' или 'all'
    - limit: максимален брой резултати (1-100)
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/repos/{owner}/{repo}/issues",
            headers=github_headers(),
            params={"state": state, "per_page": min(limit, 100)}
        )
        response.raise_for_status()
        issues = response.json()
        return [
            {
                "number": issue["number"],
                "title": issue["title"],
                "state": issue["state"],
                "author": issue["user"]["login"],
                "created_at": issue["created_at"],
                "labels": [l["name"] for l in issue["labels"]]
            }
            for issue in issues
        ]

@mcp.tool()
async def search_code(
    query: str,
    language: str = "",
    limit: int = 5
) -> list[dict[str, Any]]:
    """Търсене на код в GitHub.
    Параметри:
    - query: текст за търсене в код
    - language: филтър по език (напр. 'python', 'typescript')
    - limit: максимален брой резултати
    """
    search_query = query
    if language:
        search_query += f" language:{language}"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/search/code",
            headers=github_headers(),
            params={"q": search_query, "per_page": min(limit, 30)}
        )
        response.raise_for_status()
        data = response.json()
        return [
            {
                "repository": item["repository"]["full_name"],
                "path": item["path"],
                "url": item["html_url"]
            }
            for item in data.get("items", [])
        ]

if __name__ == "__main__":
    mcp.run(transport="stdio")

С този подход можете бързо да обвиете практически всеки REST API в MCP сървър — от Jira и Slack до вътрешни бизнес системи. Красотата е, че веднъж написан, сървърът работи с всеки MCP-съвместим клиент.

Транспортни механизми: stdio срещу Streamable HTTP

MCP поддържа два основни транспортни механизма и изборът между тях зависи от контекста на използване.

stdio (Standard Input/Output)

• Идеален за локална разработка и тестване
• Комуникацията е синхронна и бърза
• Не изисква мрежова конфигурация или автентикация
• Личните ви данни за достъп остават на вашата машина
• Важно: Никога не пишете в stdout в stdio-базирани сървъри — това ще повреди JSON-RPC съобщенията (попадал съм на този бъг повече пъти, отколкото бих искал да призная)

Streamable HTTP

• Препоръчителен за продукционна среда
• Използва единична HTTP крайна точка за GET и POST заявки
• Поддържа SSE (Server-Sent Events) за стрийминг на отговори
• Управление на сесии, възстановяване при прекъсване и OAuth интеграция
• Замени остарелия HTTP+SSE транспорт от март 2025 г.

Ето как да преминете от stdio към Streamable HTTP:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "Production Server",
    json_response=True,
    stateless_http=True  # Оптимално за мащабиране с множество работници
)

@mcp.tool()
def process_data(input_text: str) -> str:
    """Обработка на текстови данни"""
    return f"Обработен: {input_text.upper()}"

if __name__ == "__main__":
    # Streamable HTTP транспорт за продукционна среда
    mcp.run(transport="streamable-http")

За стартиране с множество работници използвайте uvicorn:

# Стартиране с 4 работника за паралелна обработка
uvicorn server:mcp --host 0.0.0.0 --port 8000 --workers 4

Автентикация и сигурност за продукция

Когато разгръщате MCP сървър в продукционна среда, сигурността е абсолютно критична. Не подценявайте този аспект — MCP сървърите имат реален достъп до системи: бази данни, API-та, файлови системи. Ето основните подходи за защита.

OAuth 2.1 автентикация (препоръчително)

Официалният MCP SDK предоставя пълна поддръжка на OAuth 2.1:

from pydantic import AnyHttpUrl
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP

class JWTTokenVerifier(TokenVerifier):
    """Верификатор на JWT токени"""
    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            # Декодиране и валидиране на JWT токена
            import jwt
            payload = jwt.decode(
                token,
                key="your-secret-key",
                algorithms=["HS256"]
            )
            return AccessToken(
                token=token,
                client_id=payload.get("client_id", ""),
                scopes=payload.get("scopes", []),
                expires_at=payload.get("exp")
            )
        except jwt.InvalidTokenError:
            return None

mcp = FastMCP(
    "Secure API Server",
    json_response=True,
    token_verifier=JWTTokenVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),
        resource_server_url=AnyHttpUrl("https://mcp.example.com"),
        required_scopes=["read", "tools"],
    ),
)

@mcp.tool()
async def get_sensitive_data(query: str) -> dict:
    """Достъп до чувствителни данни (изисква автентикация)"""
    return {"data": f"Резултат за: {query}"}

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Добри практики за сигурност

Следвайте OWASP MCP Top 10 препоръките — ето най-важните:

1. Принцип на минималните привилегии — давайте на всеки сървър минимално необходимия достъп. Ако сървърът трябва само да чете от базата данни — използвайте read-only потребител. Без изключения
2. Използвайте secrets manager — никога не записвайте API ключове в MCP конфигурацията. Използвайте HashiCorp Vault, AWS Secrets Manager или нещо подобно
3. Стартирайте сървъри в изолирана среда — контейнери или sandbox-и за сървъри с висок риск (файлова система, терминал, бази данни)
4. Логвайте всяко действие — записвайте кой клиент е извикал кой инструмент с какви аргументи. Ще ви трябва при дебъгване
5. Защита от prompt injection — валидирайте входните данни и не позволявайте на потребителски вход да манипулира поведението на инструментите

Разгръщане с Nginx и TLS

За продукционно разгръщане зад обратен прокси, Streamable HTTP транспортът изисква специфична Nginx конфигурация заради SSE стрийминга. Ето конфигурация, която работи:

server {
    listen 443 ssl;
    server_name mcp.example.com;

    ssl_certificate /etc/ssl/certs/mcp.crt;
    ssl_certificate_key /etc/ssl/private/mcp.key;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Connection "";

        # Критично за SSE стрийминг
        proxy_buffering off;
        proxy_cache off;

        # Таймаути за дълготрайни SSE връзки
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;

        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Двете най-важни настройки тук са proxy_buffering off и proxy_cache off — без тях SSE стриймингът просто няма да работи коректно.

Кога да използвате MCP и кога — не

MCP не е универсален отговор на всичко. Ето кратък и честен справочник.

Използвайте MCP когато:

• Основният потребител на интеграцията е AI модел или агент
• Имате 3 или повече интеграции, които трябва да работят заедно
• Искате да използвате един сървър с множество AI клиенти (Claude, ChatGPT, Cursor)
• Изграждате автономни AI агенти, които трябва динамично да избират инструменти

Не използвайте MCP когато:

• Нуждаете се от обикновен уеб API за не-AI приложения — REST/GraphQL си е достатъчен
• Имате единична, проста интеграция — директно API извикване е по-лесно и по-бързо
• Приложението ви не включва LLM компонент изобщо

Екосистема и готови MCP сървъри

Добрата новина е, че не е нужно да пишете всичко от нулата. Към 2026 г. вече съществуват стотици готови MCP сървъри за най-популярните услуги:

• Бази данни: PostgreSQL, Supabase, DynamoDB, MindsDB
• Код и DevOps: GitHub, Git, Docker, Azure DevOps
• Комуникация: Slack, Gmail, Microsoft Teams
• Облак: AWS, Azure, Google Cloud
• Бизнес: Stripe, Jira, Salesforce, HubSpot

Можете да разгледате пълния регистър на mcpservers.org или в официалното GitHub хранилище на MCP. Преди да пишете свой сървър, проверете дали вече не съществува такъв — често има.

Често задавани въпроси

Каква е разликата между MCP сървър и REST API?

REST API е предназначен за общо потребление от уеб и мобилни приложения. MCP сървърът е специално проектиран за AI модели — предоставя структурирани описания на инструментите, метаданни и контекст, които помагат на модела да реши кога и как да ги използва. Ако основният потребител е AI — изберете MCP. За традиционно приложение — REST/GraphQL.

Мога ли да използвам MCP с OpenAI, Claude и други модели едновременно?

Да, и точно в това е цялата идея. Един MCP сървър работи с всеки съвместим клиент — Claude Desktop, ChatGPT, Cursor, VS Code Copilot, IBM BeeAI и много други. Изграждате сървъра веднъж и го използвате навсякъде. OpenAI официално приеха MCP през март 2025 г., а Google DeepMind последваха малко след това.

Безопасно ли е да давам на AI агент достъп до базата ми данни чрез MCP?

С правилните предпазни мерки — да. Използвайте read-only потребител за базата данни, валидирайте заявките (само SELECT), блокирайте опасни ключови думи, логвайте всяко действие и стартирайте сървъра в изолирана среда. За продукция задължително добавете OAuth автентикация и TLS криптиране. Между другото, Supabase препоръчва да използвате MCP само с разработъчни (не продукционни) бази данни — добър съвет за начало.

Какъв е препоръчителният транспорт за продукционна среда?

Streamable HTTP — без въпрос. Той е стандартът за продукционни разгръщания от март 2025 г. Поддържа управление на сесии, възстановяване при прекъсване, OAuth интеграция и мащабиране с множество работници чрез uvicorn. Stdio е подходящ само за локално тестване.

Какви са текущите ограничения на MCP през 2026 г.?

Въпреки широкото приемане, MCP все още се развива активно. Основните предизвикателства: сигурностните практики все още се стандартизират (следете OWASP MCP Top 10), не всички AI клиенти поддържат всички MCP функции еднакво, и няма официален регистър за верификация на сървъри — внимавайте за подправени такива. Пътната карта за 2026 г. се фокусира върху корпоративна готовност: одитни следи, SSO интеграция и конфигурационна преносимост.