Jak zbudowac serwer MCP w Pythonie dla portalu pracy - kompletny przewodnik 2026

Wyobraź sobie że użytkownik siada do Claude'a i pisze: "Znajdź mi zdalne oferty dla React developera powyżej 20 000 PLN i powiedz mi jak wyglądają widełki w porównaniu do rynku." Zamiast odpowiedzi opartej na wiedzy z treningu - Claude wywołuje Twoje API, pobiera aktualne dane z bazy, analizuje je i odpowiada konkretnymi liczbami z dziś, nie sprzed roku. To nie jest fikcja. To jest MCP w praktyce - i możesz to zbudować w kilka godzin.

Model Context Protocol (MCP) to otwarty standard stworzony przez Anthropic, który definiuje jak modele AI mogą bezpiecznie komunikować się z zewnętrznymi narzędziami, bazami danych i API. Opublikowany w listopadzie 2024 roku, w ciągu kilku miesięcy stał się jednym z najszybciej adoptowanych standardów w ekosystemie AI. Obsługują go Claude Desktop, Cursor, Windsurf, VS Code z odpowiednim pluginem i dziesiątki innych narzędzi deweloperskich.

Ten artykuł to kompletny, praktyczny przewodnik jak zbudować własny serwer MCP dla portalu pracy - krok po kroku, od instalacji do produkcyjnego deploymentu. Użyjemy Pythona i oficjalnego SDK. Napiszemy prawdziwy kod - z obsługą błędów, walidacją, cache'owaniem, bezpieczeństwem i testami. Na końcu będziesz miał działający serwer MCP który agent AI może odpytywać o oferty pracy, wynagrodzenia, trendy technologiczne i dane rynkowe.

Jeśli prowadzisz portal pracy, agregator ofert, platformę rekrutacyjną - albo po prostu chcesz zrozumieć MCP od środka przez budowanie czegoś realnego - jesteś we właściwym miejscu.

Czym jest Model Context Protocol i dlaczego zmienia zasady gry

Żeby zrozumieć dlaczego MCP jest ważny, wróćmy do fundamentalnego problemu który rozwiązuje. Modele językowe jak Claude, GPT-4 czy Gemini wiedzą dużo - ale wiedzą to co było w ich danych treningowych. Mają datę graniczną wiedzy (knowledge cutoff). Nie wiedzą co się działo po tej dacie. Nie wiedzą co jest na Twoim serwerze. Nie wiedzą co jest w Twojej bazie danych.

Przez lata programiści obchodzili ten problem przez kopiowanie danych do kontekstu - wklejasz wyniki zapytania SQL do chatu, model to analizuje. Działa, ale ma ogromne ograniczenia: okno kontekstu ma limit, kopiowanie jest podatne na błędy, dane są aktualne tylko w momencie wklejenia. I przede wszystkim - to jest ręczne. Nie skaluje się.

Pojawiły się też Function Calling i Tool Use - mechanizmy w API pozwalające modelowi "wywoływać funkcje". Ale każda implementacja była inna. OpenAI miał swój format, Anthropic swój, Google swój. Każdy developer pisał własną integrację od zera. Ekosystem był fragmentaryczny.

MCP rozwiązuje ten problem przez standaryzację. Definiuje jeden protokół który działa dla wszystkich modeli i wszystkich narzędzi. Jeśli Twój serwer mówi MCP - działa z każdym klientem który rozumie MCP. Budujesz raz, integrujesz wszędzie.

Trzy typy zasobów w MCP

MCP definiuje trzy fundamentalne typy tego co serwer może udostępniać:

Tools - narzędzia które model może wywołać. To funkcje z nazwą, opisem i schematem parametrów. Model widzi listę dostępnych narzędzi i decyduje kiedy i które wywołać. Przykłady: search_jobs(keyword, location, salary_min), get_salary_report(technology), find_remote_positions(role). Narzędzia mogą mieć efekty uboczne - mogą pisać do bazy, wysyłać emaile, wykonywać operacje.

Resources - zasoby do odczytu. Statyczne lub dynamiczne dane które model może przeczytać. Nie wywołuje ich jako funkcji - dostaje do nich dostęp jak do pliku. Przykłady: dokumentacja API, aktualny status systemu, lista kategorii, dane konfiguracyjne.

Prompts - gotowe szablony promptów. Model lub klient może je wybrać i użyć jako punkt startowy dla konwersacji. Przydatne gdy chcesz zdefiniować standardowe scenariusze użycia Twojego serwera.

W praktyce dla większości portali pracy najważniejsze są Tools. Resources przydają się do dokumentacji i stałych danych. Prompts to bonus który możesz dodać później.

Jak wygląda wywołanie narzędzia MCP

Przepływ jest prosty. Użytkownik pisze pytanie do Claude'a (lub innego klienta MCP). Model analizuje pytanie i decyduje że potrzebuje wywołać narzędzie. Wysyła do serwera MCP request z nazwą narzędzia i parametrami. Serwer wykonuje operację i zwraca wynik. Model otrzymuje wynik i formułuje odpowiedź dla użytkownika.

Wszystko dzieje się transparentnie - użytkownik widzi tylko pytanie i odpowiedź. Wywołania narzędzi są ukryte, chyba że klient (np. Claude Desktop) zdecyduje się je pokazać.

Transport: stdio vs HTTP

MCP definiuje dwa sposoby transportu komunikacji między klientem a serwerem.

stdio (standard input/output) - serwer działa jako lokalny proces. Klient (np. Claude Desktop) uruchamia serwer jako subprocess i komunikuje się przez standardowe wejście i wyjście. To najprostszy setup - idealne do lokalnego developmentu i narzędzi developerskich.

HTTP z SSE (Server-Sent Events) - serwer działa jako endpoint HTTP. Klient łączy się przez sieć. To jest właściwy transport dla deploymentu produkcyjnego - serwer może być na osobnym serwerze, może obsługiwać wielu klientów jednocześnie, można go skalować.

Zaczniemy od stdio (lokalny development), a potem pokażemy jak przepiąć na HTTP dla produkcji.

Wymagania i konfiguracja środowiska

Zanim napiszemy pierwszy kod, upewniamy się że mamy właściwe środowisko. Potrzebujemy Pythona 3.10 lub nowszego - MCP SDK używa nowszych funkcji typowania które nie działają na starszych wersjach.

python3 --version
# Python 3.12.x lub nowszy - idealne
# Python 3.10.x - minimalne wymaganie

# Tworzymy wirtualne środowisko
python3 -m venv venv-mcp
source venv-mcp/bin/activate  # Linux/Mac
# lub
.\venv-mcp\Scripts\activate   # Windows

# Instalujemy zależności
pip install mcp httpx pydantic python-dotenv

# Dla cache'owania
pip install redis

# Dla testów
pip install pytest pytest-asyncio httpx

Plik requirements.txt który będziemy rozbudowywać:

mcp>=1.0.0
httpx>=0.27.0
pydantic>=2.0.0
python-dotenv>=1.0.0
redis>=5.0.0
pytest>=8.0.0
pytest-asyncio>=0.23.0

Struktura projektu

Zanim napiszemy kod, planujemy strukturę plików. Dobra struktura od początku oszczędza czasu przy skalowaniu.

jobs-mcp-server/
├── server.py              # Główny plik serwera MCP
├── tools/
│   ├── __init__.py
│   ├── search.py          # Narzędzia wyszukiwania
│   ├── salary.py          # Narzędzia wynagrodzeń
│   ├── analytics.py       # Narzędzia analityki
│   └── market.py          # Narzędzia trendów rynkowych
├── resources/
│   ├── __init__.py
│   └── documentation.py   # Zasoby dokumentacji
├── prompts/
│   ├── __init__.py
│   └── templates.py       # Szablony promptów
├── api/
│   ├── __init__.py
│   └── client.py          # Klient do backend API
├── cache/
│   ├── __init__.py
│   └── redis_cache.py     # Warstwa cache'owania
├── security/
│   ├── __init__.py
│   └── validators.py      # Walidacja inputu
├── tests/
│   ├── test_search.py
│   ├── test_salary.py
│   └── test_security.py
├── .env.example
├── requirements.txt
└── README.md

Pierwszy serwer MCP - od Hello World do struktury produkcyjnej

Zacznijmy od absolutnego minimum - działający serwer który Claude może wykryć i wywołać. Potem będziemy go rozbudowywać.

Minimalny serwer - Hello World

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("2HR Jobs MCP Server")

@mcp.tool()
def hello() -> str:
    """Testowe narzędzie potwierdzające że serwer działa."""
    return "Serwer MCP dla portalu pracy 2HR działa poprawnie!"

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

To jest naprawdę minimum. FastMCP to wysokopoziomowa warstwa SDK która ukrywa większość złożoności protokołu. Dekorator @mcp.tool() rejestruje funkcję jako narzędzie dostępne dla modelu. Docstring funkcji staje się opisem narzędzia - model AI czyta ten opis żeby wiedzieć kiedy narzędzie wywołać.

Uruchamiamy:

python server.py
# Serwer uruchamia się i czeka na połączenia przez stdio

Dodajemy typowanie i lepsze opisy

Model AI używa opisów narzędzi i ich parametrów żeby wiedzieć jak je wywołać. Im lepszy opis, tym trafniejsze wywołania. Rozbudujmy nasz serwer z właściwym typowaniem:

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

mcp = FastMCP(
    name="2HR Jobs MCP Server",
    version="1.0.0",
)

class JobSearchParams(BaseModel):
    keyword: str = Field(description="Słowo kluczowe - technologia, rola lub umiejętność")
    location: Optional[str] = Field(default=None, description="Miasto lub 'remote' dla pracy zdalnej")
    salary_min: Optional[int] = Field(default=None, description="Minimalne wynagrodzenie w PLN brutto miesięcznie")
    salary_max: Optional[int] = Field(default=None, description="Maksymalne wynagrodzenie w PLN brutto miesięcznie")
    experience_level: Optional[str] = Field(
        default=None,
        description="Poziom doświadczenia: junior, mid, senior, lead"
    )
    employment_type: Optional[str] = Field(
        default=None,
        description="Typ zatrudnienia: b2b, uop, zlecenie"
    )
    limit: int = Field(default=10, ge=1, le=50, description="Maksymalna liczba wyników (1-50)")

@mcp.tool()
def search_jobs(params: JobSearchParams) -> dict:
    """
    Wyszukuje oferty pracy IT w Polsce na podstawie podanych kryteriów.
    Użyj tego narzędzia gdy użytkownik pyta o dostępne oferty pracy,
    stanowiska, lub chce znaleźć pracę w określonej technologii lub mieście.
    """
    # Tu będzie prawdziwa implementacja
    return {
        "status": "ok",
        "results": [],
        "query": params.model_dump()
    }

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

Kilka ważnych rzeczy w tym kodzie. Po pierwsze - Pydantic do walidacji parametrów. SDK MCP automatycznie waliduje parametry przed wywołaniem narzędzia. Jeśli ktoś (lub model) poda nieprawidłowy limit (np. 1000), dostanie błąd walidacji zanim narzędzie zostanie wywołane. Po drugie - Field z description. Te opisy trafiają do schematu narzędzia który model widzi. Im bardziej precyzyjny opis, tym lepiej model rozumie jak używać parametru. Po trzecie - docstring narzędzia. To najważniejszy opis - model używa go do decyzji kiedy wywołać narzędzie. Powinien opisywać co narzędzie robi i w jakich sytuacjach jest przydatne.

Klient API - warstwa komunikacji z backendem

Serwer MCP nie jest bazą danych - jest warstwą pośrednią między modelem AI a Twoim istniejącym backend API. W produkcji serwer MCP wysyła requesty do Twojego REST API, które odpytuje bazę danych.

# api/client.py
import httpx
import os
from typing import Optional, Any
from dotenv import load_dotenv

load_dotenv()

class JobsApiClient:
    """Klient do komunikacji z backend API portalu pracy."""

    def __init__(self):
        self.base_url = os.getenv("JOBS_API_URL", "https://api.2hr.pl/v1")
        self.api_key = os.getenv("JOBS_API_KEY", "")
        self.timeout = float(os.getenv("API_TIMEOUT", "10.0"))

    def _get_headers(self) -> dict:
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "2HR-MCP-Server/1.0",
        }
        if self.api_key:
            headers["X-API-Key"] = self.api_key
        return headers

    async def get(self, endpoint: str, params: Optional[dict] = None) -> Any:
        """Wykonuje GET request do API."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.base_url}/{endpoint.lstrip('/')}",
                params=params,
                headers=self._get_headers()
            )
            response.raise_for_status()
            return response.json()

    async def search_jobs(
        self,
        keyword: str,
        location: Optional[str] = None,
        salary_min: Optional[int] = None,
        salary_max: Optional[int] = None,
        experience_level: Optional[str] = None,
        employment_type: Optional[str] = None,
        limit: int = 10
    ) -> dict:
        """Wyszukuje oferty pracy przez API."""
        params = {
            "q": keyword,
            "limit": limit,
        }
        if location:
            params["location"] = location
        if salary_min:
            params["salary_min"] = salary_min
        if salary_max:
            params["salary_max"] = salary_max
        if experience_level:
            params["level"] = experience_level
        if employment_type:
            params["type"] = employment_type

        return await self.get("/jobs/search", params)

    async def get_salary_report(self, technology: str) -> dict:
        """Pobiera raport wynagrodzeń dla technologii."""
        return await self.get(f"/salary/report", {"technology": technology})

    async def get_top_technologies(self, limit: int = 20) -> dict:
        """Pobiera ranking najpopularniejszych technologii."""
        return await self.get("/analytics/top-technologies", {"limit": limit})

    async def compare_roles(self, role_a: str, role_b: str) -> dict:
        """Porównuje dwie role zawodowe."""
        return await self.get("/analytics/compare", {"role_a": role_a, "role_b": role_b})

    async def get_market_trends(self, period: str = "3m") -> dict:
        """Pobiera trendy rynkowe z ostatnich miesięcy."""
        return await self.get("/analytics/trends", {"period": period})

    async def get_remote_jobs(self, keyword: Optional[str] = None, limit: int = 10) -> dict:
        """Pobiera oferty pracy zdalnej."""
        params = {"remote": "true", "limit": limit}
        if keyword:
            params["q"] = keyword
        return await self.get("/jobs/search", params)

    async def get_jobs_by_city(self, city: str, keyword: Optional[str] = None, limit: int = 10) -> dict:
        """Pobiera oferty pracy w konkretnym mieście."""
        params = {"location": city, "limit": limit}
        if keyword:
            params["q"] = keyword
        return await self.get("/jobs/search", params)

Plik .env.example który developer powinien skopiować do .env:

JOBS_API_URL=https://api.twoja-domena.pl/v1
JOBS_API_KEY=twoj_klucz_api
API_TIMEOUT=10.0
REDIS_URL=redis://localhost:6379/0
CACHE_TTL_JOBS=300
CACHE_TTL_SALARY=3600
CACHE_TTL_ANALYTICS=1800
LOG_LEVEL=INFO

Warstwa cache - Redis dla wydajności

Serwer MCP może być wywołany setki razy dziennie przez agentów AI. Bez cache każde wywołanie to request do Twojego API. Z cache - większość odpowiedzi serwujemy bezpośrednio z pamięci. Dla danych które zmieniają się rzadko (wynagrodzenia, trendy) cache na godzinę to ogromna oszczędność zasobów.

# cache/redis_cache.py
import json
import hashlib
import redis.asyncio as aioredis
import os
from typing import Optional, Any

class RedisCache:
    """Warstwa cache'owania oparta na Redis."""

    def __init__(self):
        self.redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
        self._client: Optional[aioredis.Redis] = None

    async def get_client(self) -> aioredis.Redis:
        if self._client is None:
            self._client = await aioredis.from_url(
                self.redis_url,
                encoding="utf-8",
                decode_responses=True
            )
        return self._client

    def _make_key(self, prefix: str, params: dict) -> str:
        """Generuje klucz cache na podstawie parametrów."""
        params_hash = hashlib.md5(
            json.dumps(params, sort_keys=True, ensure_ascii=False).encode()
        ).hexdigest()
        return f"mcp:{prefix}:{params_hash}"

    async def get(self, key: str) -> Optional[Any]:
        """Pobiera wartość z cache. Zwraca None jeśli nie istnieje."""
        try:
            client = await self.get_client()
            value = await client.get(key)
            if value:
                return json.loads(value)
            return None
        except Exception:
            return None  # Cache miss na błędzie - nie blokujemy aplikacji

    async def set(self, key: str, value: Any, ttl: int = 300) -> None:
        """Zapisuje wartość do cache z TTL w sekundach."""
        try:
            client = await self.get_client()
            await client.setex(key, ttl, json.dumps(value, ensure_ascii=False))
        except Exception:
            pass  # Błąd cache nie przerywa działania serwera

    async def get_or_set(
        self,
        prefix: str,
        params: dict,
        fetch_fn,
        ttl: int = 300
    ) -> Any:
        """
        Wzorzec cache-aside:
        1. Sprawdź cache
        2. Jeśli miss - wywołaj fetch_fn
        3. Zapisz wynik do cache
        4. Zwróć wynik
        """
        key = self._make_key(prefix, params)
        cached = await self.get(key)

        if cached is not None:
            return {**cached, "_cached": True}

        result = await fetch_fn()
        await self.set(key, result, ttl)
        return {**result, "_cached": False}

    async def invalidate(self, prefix: str) -> int:
        """Usuwa wszystkie klucze z danym prefixem."""
        try:
            client = await self.get_client()
            pattern = f"mcp:{prefix}:*"
            keys = await client.keys(pattern)
            if keys:
                return await client.delete(*keys)
            return 0
        except Exception:
            return 0

# Singleton
_cache: Optional[RedisCache] = None

def get_cache() -> RedisCache:
    global _cache
    if _cache is None:
        _cache = RedisCache()
    return _cache

Narzędzia wyszukiwania ofert - pełna implementacja

Teraz implementujemy prawdziwe narzędzia. To jest serce serwera MCP - funkcje które model AI będzie wywoływał najczęściej.

# tools/search.py
import asyncio
from typing import Optional
from pydantic import BaseModel, Field, field_validator
from mcp.server.fastmcp import FastMCP
from api.client import JobsApiClient
from cache.redis_cache import get_cache
from security.validators import sanitize_search_input
import os

# TTL dla różnych typów danych
TTL_JOBS = int(os.getenv("CACHE_TTL_JOBS", "300"))          # 5 minut
TTL_REMOTE = int(os.getenv("CACHE_TTL_REMOTE", "300"))      # 5 minut
TTL_CITY = int(os.getenv("CACHE_TTL_CITY", "600"))          # 10 minut

api_client = JobsApiClient()
cache = get_cache()

class JobSearchInput(BaseModel):
    keyword: str = Field(
        description="Technologia, rola lub umiejętność (np. 'Python', 'React Developer', 'DevOps')"
    )
    location: Optional[str] = Field(
        default=None,
        description="Miasto (np. 'Warszawa', 'Kraków', 'Gdańsk') lub 'remote' dla pracy zdalnej"
    )
    salary_min: Optional[int] = Field(
        default=None,
        ge=1000,
        le=200000,
        description="Minimalne wynagrodzenie w PLN brutto/miesiąc"
    )
    salary_max: Optional[int] = Field(
        default=None,
        ge=1000,
        le=200000,
        description="Maksymalne wynagrodzenie w PLN brutto/miesiąc"
    )
    experience_level: Optional[str] = Field(
        default=None,
        description="Poziom: 'junior' (0-2 lata), 'mid' (2-5 lat), 'senior' (5+ lat), 'lead'"
    )
    employment_type: Optional[str] = Field(
        default=None,
        description="Forma zatrudnienia: 'b2b' (faktura), 'uop' (umowa o pracę), 'zlecenie'"
    )
    limit: int = Field(
        default=10,
        ge=1,
        le=50,
        description="Liczba wyników do zwrócenia (domyślnie 10, maksymalnie 50)"
    )

    @field_validator('keyword')
    @classmethod
    def validate_keyword(cls, v: str) -> str:
        v = sanitize_search_input(v)
        if len(v) < 1:
            raise ValueError('Słowo kluczowe nie może być puste')
        if len(v) > 100:
            raise ValueError('Słowo kluczowe jest za długie (max 100 znaków)')
        return v

    @field_validator('experience_level')
    @classmethod
    def validate_experience(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        allowed = {'junior', 'mid', 'senior', 'lead', 'principal'}
        if v.lower() not in allowed:
            raise ValueError(f'Poziom musi być jednym z: {", ".join(allowed)}')
        return v.lower()

    @field_validator('employment_type')
    @classmethod
    def validate_employment(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        allowed = {'b2b', 'uop', 'zlecenie', 'umowa_o_prace'}
        if v.lower() not in allowed:
            raise ValueError(f'Typ zatrudnienia musi być jednym z: {", ".join(allowed)}')
        return v.lower()

def format_job_for_ai(job: dict) -> dict:
    """Formatuje dane oferty pracy dla optymalnej prezentacji przez AI."""
    salary_info = "nie podano"
    if job.get("salary_min") and job.get("salary_max"):
        salary_info = f"{job['salary_min']:,} - {job['salary_max']:,} PLN {job.get('employment_type', '').upper()}"
    elif job.get("salary_min"):
        salary_info = f"od {job['salary_min']:,} PLN"

    return {
        "id": job.get("id"),
        "tytuł": job.get("title", ""),
        "firma": job.get("company", ""),
        "lokalizacja": job.get("location", ""),
        "zdalna": job.get("is_remote", False),
        "wynagrodzenie": salary_info,
        "poziom": job.get("experience_level", ""),
        "technologie": job.get("technologies", []),
        "data_dodania": job.get("created_at", ""),
        "url": job.get("url", ""),
        "opis_krotki": job.get("snippet", "")[:300] if job.get("snippet") else ""
    }

def register_search_tools(mcp: FastMCP) -> None:
    """Rejestruje wszystkie narzędzia wyszukiwania w serwerze MCP."""

    @mcp.tool()
    async def search_jobs(params: JobSearchInput) -> dict:
        """
        Wyszukuje oferty pracy IT w Polsce.

        Użyj gdy użytkownik:
        - Pyta o dostępne oferty pracy dla konkretnej technologii
        - Chce znaleźć pracę w konkretnym mieście
        - Szuka ofert z określonym wynagrodzeniem
        - Pyta "czy są oferty dla X developera?"
        - Chce porównać dostępność ofert dla różnych technologii

        Przykłady dobrych wywołań:
        - search_jobs(keyword="Python", location="Warszawa")
        - search_jobs(keyword="React", salary_min=15000, experience_level="senior")
        - search_jobs(keyword="DevOps", location="remote", employment_type="b2b")
        """
        cache_params = params.model_dump()

        async def fetch():
            return await api_client.search_jobs(
                keyword=params.keyword,
                location=params.location,
                salary_min=params.salary_min,
                salary_max=params.salary_max,
                experience_level=params.experience_level,
                employment_type=params.employment_type,
                limit=params.limit
            )

        result = await cache.get_or_set("jobs_search", cache_params, fetch, TTL_JOBS)

        jobs = result.get("jobs", [])
        formatted_jobs = [format_job_for_ai(job) for job in jobs]

        return {
            "zapytanie": {
                "słowo_kluczowe": params.keyword,
                "lokalizacja": params.location or "wszystkie",
                "wynagrodzenie_min": params.salary_min,
                "wynagrodzenie_max": params.salary_max,
                "poziom": params.experience_level,
                "typ_zatrudnienia": params.employment_type,
            },
            "liczba_wyników": result.get("total", len(formatted_jobs)),
            "oferty": formatted_jobs,
            "czy_z_cache": result.get("_cached", False),
        }

    @mcp.tool()
    async def find_remote_jobs(
        keyword: Optional[str] = None,
        salary_min: Optional[int] = None,
        limit: int = 10
    ) -> dict:
        """
        Wyszukuje wyłącznie oferty pracy zdalnej (remote/home office).

        Użyj gdy użytkownik pyta o:
        - Pracę zdalną w IT
        - Oferty home office
        - Zdalne stanowiska dla konkretnej technologii
        - "Czy można pracować zdalnie jako X?"

        Zwraca tylko oferty z możliwością pracy zdalnej.
        """
        if keyword:
            keyword = sanitize_search_input(keyword)

        cache_params = {"keyword": keyword, "salary_min": salary_min, "limit": limit}

        async def fetch():
            return await api_client.get_remote_jobs(keyword=keyword, limit=limit)

        result = await cache.get_or_set("remote_jobs", cache_params, fetch, TTL_REMOTE)
        jobs = result.get("jobs", [])

        if salary_min:
            jobs = [
                j for j in jobs
                if j.get("salary_min") and j.get("salary_min") >= salary_min
            ]

        return {
            "informacja": "Wszystkie poniższe oferty umożliwiają pracę zdalną",
            "filtr_wynagrodzenia_min": salary_min,
            "liczba_wyników": len(jobs),
            "oferty": [format_job_for_ai(job) for job in jobs],
        }

    @mcp.tool()
    async def find_jobs_in_city(
        city: str,
        keyword: Optional[str] = None,
        salary_min: Optional[int] = None,
        limit: int = 10
    ) -> dict:
        """
        Wyszukuje oferty pracy w konkretnym polskim mieście.

        Użyj gdy użytkownik pyta o:
        - Oferty w konkretnej lokalizacji
        - "Ile jest ofert w Krakowie dla Javy?"
        - Rynek pracy w konkretnym mieście
        - Porównanie ofert w różnych miastach (wywołaj kilka razy)

        Obsługuje miasta: Warszawa, Kraków, Wrocław, Gdańsk, Gdynia, Poznań,
        Łódź, Katowice, Rzeszów, Szczecin i inne.
        """
        city = sanitize_search_input(city)
        if keyword:
            keyword = sanitize_search_input(keyword)

        cache_params = {"city": city, "keyword": keyword, "salary_min": salary_min, "limit": limit}

        async def fetch():
            return await api_client.get_jobs_by_city(city=city, keyword=keyword, limit=limit)

        result = await cache.get_or_set("city_jobs", cache_params, fetch, TTL_CITY)
        jobs = result.get("jobs", [])

        if salary_min:
            jobs = [j for j in jobs if j.get("salary_min") and j.get("salary_min") >= salary_min]

        return {
            "miasto": city,
            "technologia": keyword or "wszystkie",
            "liczba_ofert": len(jobs),
            "oferty": [format_job_for_ai(job) for job in jobs],
        }

    @mcp.tool()
    async def get_latest_jobs(limit: int = 20) -> dict:
        """
        Pobiera najnowsze oferty pracy IT dodane do portalu.

        Użyj gdy użytkownik pyta o:
        - Najnowsze oferty
        - Co nowego pojawiło się na rynku
        - Świeże oferty z ostatnich dni
        """
        cache_params = {"limit": limit}

        async def fetch():
            return await api_client.get("/jobs/latest", {"limit": limit, "sort": "date_desc"})

        result = await cache.get_or_set("latest_jobs", cache_params, fetch, 120)  # 2 minuty - świeże dane
        jobs = result.get("jobs", [])

        return {
            "informacja": "Najnowsze oferty pracy IT w Polsce",
            "liczba": len(jobs),
            "oferty": [format_job_for_ai(job) for job in jobs],
        }

Narzędzia analizy wynagrodzeń

Wynagrodzenia to jeden z najczęściej szukanych tematów w IT. Użytkownicy pytają modele AI o widełki bo nie wiedzą czy aktualna oferta jest uczciwa, bo negocjują podwyżkę, bo planują zmianę pracy. Serwer MCP z danymi wynagrodzeń to ogromna wartość.

# tools/salary.py
from typing import Optional
from pydantic import BaseModel, Field, field_validator
from mcp.server.fastmcp import FastMCP
from api.client import JobsApiClient
from cache.redis_cache import get_cache
from security.validators import sanitize_search_input
import os

TTL_SALARY = int(os.getenv("CACHE_TTL_SALARY", "3600"))   # 1 godzina

api_client = JobsApiClient()
cache = get_cache()

SUPPORTED_TECHNOLOGIES = {
    "python", "javascript", "typescript", "java", "php", "go", "rust",
    "csharp", "c#", "cpp", "c++", "ruby", "scala", "kotlin", "swift",
    "react", "angular", "vue", "node", "nodejs", "django", "fastapi",
    "spring", "laravel", "symfony", "flutter", "react native",
    "aws", "azure", "gcp", "docker", "kubernetes", "terraform",
    "devops", "sre", "fullstack", "backend", "frontend", "mobile",
    "data scientist", "data engineer", "machine learning", "ai engineer",
    "cybersecurity", "security", "qa", "tester", "product manager",
    "scrum master", "project manager", "ux designer", "ui designer"
}

def register_salary_tools(mcp: FastMCP) -> None:

    @mcp.tool()
    async def get_salary_report(
        technology: str,
        experience_level: Optional[str] = None,
        employment_type: Optional[str] = None
    ) -> dict:
        """
        Pobiera szczegółowy raport wynagrodzeń dla konkretnej technologii lub roli w Polsce.

        Użyj gdy użytkownik:
        - Pyta ile zarabia X developer
        - Chce wiedzieć czy oferta jest uczciwa finansowo
        - Negocjuje wynagrodzenie i potrzebuje danych rynkowych
        - Porównuje zarobki w różnych technologiach
        - Pyta "czy warto przejść z X na Y dla lepszych zarobków?"

        Zwraca: medianę, percentyle (p25, p75, p90), rozkład B2B vs UoP,
        porównanie miast, trend (rosnące/stabilne/spadające).
        """
        technology = sanitize_search_input(technology).lower()
        if experience_level:
            experience_level = experience_level.lower()

        cache_params = {
            "technology": technology,
            "experience_level": experience_level,
            "employment_type": employment_type
        }

        async def fetch():
            return await api_client.get_salary_report(technology)

        result = await cache.get_or_set("salary_report", cache_params, fetch, TTL_SALARY)
        data = result.get("data", {})

        # Filtrujemy po poziomie jeśli podany
        if experience_level and "by_level" in data:
            level_data = data["by_level"].get(experience_level, {})
            return {
                "technologia": technology,
                "poziom_doswiadczenia": experience_level,
                "mediana": level_data.get("median"),
                "percentyl_25": level_data.get("p25"),
                "percentyl_75": level_data.get("p75"),
                "percentyl_90": level_data.get("p90"),
                "typ_zatrudnienia": level_data.get("most_common_type"),
                "liczba_ofert_w_analizie": level_data.get("sample_size"),
                "trend": level_data.get("trend", "stabilne"),
                "waluta": "PLN",
                "okres": "brutto/miesiąc",
            }

        return {
            "technologia": technology,
            "wszystkie_poziomy": {
                "junior": data.get("junior", {}),
                "mid": data.get("mid", {}),
                "senior": data.get("senior", {}),
                "lead": data.get("lead", {}),
            },
            "według_miasta": data.get("by_city", {}),
            "b2b_vs_uop": data.get("contract_comparison", {}),
            "trend_roczny": data.get("yearly_trend", "stabilne"),
            "liczba_ofert_w_analizie": data.get("sample_size", 0),
            "ostatnia_aktualizacja": data.get("updated_at", ""),
            "waluta": "PLN",
            "okres": "brutto/miesiąc",
        }

    @mcp.tool()
    async def compare_salaries(
        technology_a: str,
        technology_b: str,
        experience_level: Optional[str] = None
    ) -> dict:
        """
        Porównuje wynagrodzenia dwóch technologii lub ról w Polsce.

        Użyj gdy użytkownik:
        - Pyta "czy warto przejść z PHP na Python ze względu na zarobki?"
        - Chce porównać widełki dla Frontend vs Backend
        - Zastanawia się która technologia lepiej płaci
        - Analizuje czy zmiana specjalizacji ma sens finansowo

        Wywołuje dane dla obu technologii równolegle dla szybkości.
        """
        technology_a = sanitize_search_input(technology_a).lower()
        technology_b = sanitize_search_input(technology_b).lower()

        # Pobieramy oba raporty równolegle
        results_a, results_b = await asyncio.gather(
            cache.get_or_set(
                "salary_report",
                {"technology": technology_a, "experience_level": experience_level, "employment_type": None},
                lambda: api_client.get_salary_report(technology_a),
                TTL_SALARY
            ),
            cache.get_or_set(
                "salary_report",
                {"technology": technology_b, "experience_level": experience_level, "employment_type": None},
                lambda: api_client.get_salary_report(technology_b),
                TTL_SALARY
            )
        )

        data_a = results_a.get("data", {})
        data_b = results_b.get("data", {})

        def get_median(data, level=None):
            if level and "by_level" in data:
                return data["by_level"].get(level, {}).get("median")
            return data.get("overall_median")

        median_a = get_median(data_a, experience_level)
        median_b = get_median(data_b, experience_level)

        different = None
        if median_a and median_b:
            diff = median_b - median_a
            diff_pct = (diff / median_a * 100) if median_a else 0
            different = {
                "różnica_PLN": diff,
                "różnica_procent": round(diff_pct, 1),
                "lepsza_technologia": technology_b if diff > 0 else technology_a,
            }

        return {
            "porównanie": f"{technology_a} vs {technology_b}",
            "poziom_doswiadczenia": experience_level or "wszystkie",
            technology_a: {
                "mediana": median_a,
                "trend": data_a.get("yearly_trend"),
                "liczba_ofert": data_a.get("sample_size"),
            },
            technology_b: {
                "mediana": median_b,
                "trend": data_b.get("yearly_trend"),
                "liczba_ofert": data_b.get("sample_size"),
            },
            "wynik_porównania": different,
            "uwaga": "Dane oparte na ofertach z portalu 2hr.pl z ostatnich 90 dni",
        }

    @mcp.tool()
    async def check_salary_fairness(
        technology: str,
        offered_salary: int,
        experience_level: str,
        employment_type: str = "b2b",
        location: Optional[str] = None
    ) -> dict:
        """
        Ocenia czy zaproponowane wynagrodzenie jest uczciwe rynkowo.

        Użyj gdy użytkownik:
        - Dostał ofertę pracy i chce wiedzieć czy pensja jest OK
        - Pyta "czy 18 000 PLN B2B dla seniora React to dobra stawka?"
        - Negocjuje wynagrodzenie i potrzebuje argumentów
        - Chce wiedzieć ile może jeszcze wynegocjować

        Zwraca ocenę (poniżej rynku / rynkowe / powyżej rynku) z uzasadnieniem.
        """
        technology = sanitize_search_input(technology).lower()
        experience_level = experience_level.lower()

        cache_params = {
            "technology": technology,
            "experience_level": experience_level,
            "employment_type": employment_type
        }

        async def fetch():
            return await api_client.get_salary_report(technology)

        result = await cache.get_or_set("salary_report", cache_params, fetch, TTL_SALARY)
        data = result.get("data", {})
        level_data = data.get("by_level", {}).get(experience_level, {})

        p25 = level_data.get("p25")
        median = level_data.get("median")
        p75 = level_data.get("p75")
        p90 = level_data.get("p90")

        assessment = "brak danych"
        recommendation = ""
        negotiation_space = None

        if median:
            if offered_salary < (p25 or median * 0.8):
                assessment = "znacznie poniżej rynku"
                recommendation = f"Rynek płaci medianę {median:,} PLN. Możesz negocjować znacznie wyżej."
                negotiation_space = median - offered_salary
            elif offered_salary < median * 0.95:
                assessment = "poniżej rynku"
                recommendation = f"Mediana dla {technology} {experience_level} to {median:,} PLN. Masz przestrzeń do negocjacji."
                negotiation_space = median - offered_salary
            elif offered_salary <= median * 1.05:
                assessment = "rynkowe"
                recommendation = "Wynagrodzenie jest zgodne z medianą rynkową."
            elif offered_salary <= (p75 or median * 1.25):
                assessment = "powyżej mediany"
                recommendation = "Dobra oferta - powyżej mediany rynkowej."
            else:
                assessment = "bardzo powyżej rynku"
                recommendation = "Wyjątkowo dobra oferta - top 25% rynku."

        return {
            "technologia": technology,
            "poziom": experience_level,
            "typ_zatrudnienia": employment_type,
            "proponowane_wynagrodzenie": offered_salary,
            "ocena": assessment,
            "rekomendacja": recommendation,
            "przestrzeń_negocjacyjna_PLN": negotiation_space,
            "dane_rynkowe": {
                "percentyl_25": p25,
                "mediana": median,
                "percentyl_75": p75,
                "percentyl_90": p90,
            },
            "źródło": "Dane z portalu 2hr.pl, ostatnie 90 dni",
        }

Narzędzia analityki rynkowej

Poza wyszukiwaniem ofert i danymi wynagrodzeń, serwer MCP może dostarczać szerszą analitykę rynku pracy - które technologie rosną, które maleją, gdzie są największe braki kadrowe, jakie umiejętności są najczęściej wymagane.

# tools/analytics.py
from typing import Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
from api.client import JobsApiClient
from cache.redis_cache import get_cache
from security.validators import sanitize_search_input
import asyncio
import os

TTL_ANALYTICS = int(os.getenv("CACHE_TTL_ANALYTICS", "1800"))   # 30 minut
TTL_TRENDS = int(os.getenv("CACHE_TTL_TRENDS", "3600"))          # 1 godzina

api_client = JobsApiClient()
cache = get_cache()

def register_analytics_tools(mcp: FastMCP) -> None:

    @mcp.tool()
    async def get_top_technologies(
        category: Optional[str] = None,
        limit: int = 20
    ) -> dict:
        """
        Pobiera ranking najpopularniejszych technologii IT w Polsce według liczby ofert pracy.

        Użyj gdy użytkownik:
        - Pyta które technologie są najczęściej wymagane
        - Chce wiedzieć co jest popularne na rynku
        - Zastanawia się czego się nauczyć dla lepszej zatrudnialności
        - Pyta "jakie technologie są teraz na topie w Polsce?"

        Kategorie: 'backend', 'frontend', 'mobile', 'devops', 'data', 'security'
        """
        cache_params = {"category": category, "limit": limit}

        async def fetch():
            params = {"limit": limit}
            if category:
                params["category"] = category
            return await api_client.get("/analytics/top-technologies", params)

        result = await cache.get_or_set("top_technologies", cache_params, fetch, TTL_ANALYTICS)
        technologies = result.get("technologies", [])

        return {
            "informacja": f"Top {limit} technologii według liczby aktywnych ofert pracy",
            "kategoria": category or "wszystkie",
            "ranking": [
                {
                    "pozycja": idx + 1,
                    "technologia": tech.get("name"),
                    "liczba_ofert": tech.get("job_count"),
                    "zmiana_miesięczna": tech.get("monthly_change"),
                    "trend": tech.get("trend", "stabilny"),
                    "mediana_wynagrodzenia": tech.get("median_salary"),
                }
                for idx, tech in enumerate(technologies)
            ],
            "data_analizy": result.get("generated_at"),
        }

    @mcp.tool()
    async def compare_roles(
        role_a: str,
        role_b: str
    ) -> dict:
        """
        Porównuje dwie role zawodowe IT pod kątem dostępności ofert, wynagrodzeń i wymagań.

        Użyj gdy użytkownik:
        - Zastanawia się między dwiema ścieżkami kariery
        - Pyta "Backend vs Frontend - co wybrać?"
        - Chce porównać DevOps z SRE, Data Scientist z Data Engineer
        - Analizuje ścieżkę kariery i chce danych do decyzji

        Zwraca pełne porównanie: liczba ofert, wynagrodzenia, wymagane skille,
        poziom trudności wejścia, perspektywy wzrostu.
        """
        role_a = sanitize_search_input(role_a)
        role_b = sanitize_search_input(role_b)

        # Pobieramy oba równolegle
        result_a, result_b = await asyncio.gather(
            cache.get_or_set(
                "role_analysis",
                {"role": role_a},
                lambda: api_client.get("/analytics/role", {"role": role_a}),
                TTL_ANALYTICS
            ),
            cache.get_or_set(
                "role_analysis",
                {"role": role_b},
                lambda: api_client.get("/analytics/role", {"role": role_b}),
                TTL_ANALYTICS
            )
        )

        def extract_role_data(result: dict, role_name: str) -> dict:
            data = result.get("data", {})
            return {
                "rola": role_name,
                "liczba_aktywnych_ofert": data.get("active_jobs", 0),
                "mediana_wynagrodzenia_senior": data.get("senior_median_salary"),
                "wymagane_technologie_top5": data.get("top_skills", [])[:5],
                "poziom_trudnosci_wejscia": data.get("entry_difficulty", ""),
                "dostepnosc_pracy_zdalnej": data.get("remote_percentage", 0),
                "trend_ofert": data.get("jobs_trend", ""),
                "czas_do_pierwszej_pracy": data.get("time_to_hire_days"),
            }

        data_a = extract_role_data(result_a, role_a)
        data_b = extract_role_data(result_b, role_b)

        winner_salary = None
        winner_jobs = None
        if data_a.get("mediana_wynagrodzenia_senior") and data_b.get("mediana_wynagrodzenia_senior"):
            winner_salary = role_a if data_a["mediana_wynagrodzenia_senior"] > data_b["mediana_wynagrodzenia_senior"] else role_b
        if data_a.get("liczba_aktywnych_ofert") and data_b.get("liczba_aktywnych_ofert"):
            winner_jobs = role_a if data_a["liczba_aktywnych_ofert"] > data_b["liczba_aktywnych_ofert"] else role_b

        return {
            "porównanie": f"{role_a} vs {role_b}",
            role_a: data_a,
            role_b: data_b,
            "podsumowanie": {
                "lepsze_wynagrodzenie": winner_salary,
                "więcej_ofert": winner_jobs,
                "uwaga": "Ostateczny wybór zależy od Twoich zainteresowań i obecnych umiejętności",
            }
        }

    @mcp.tool()
    async def get_market_trends(
        period: str = "3m",
        technology: Optional[str] = None
    ) -> dict:
        """
        Pobiera trendy rynku pracy IT w Polsce za wybrany okres.

        Użyj gdy użytkownik:
        - Pyta jak zmienia się rynek pracy IT w Polsce
        - Chce wiedzieć czy technologia X rośnie czy maleje
        - Pyta o sezonowość ofert pracy
        - Analizuje kierunek w którym zmierza branża
        - Pyta "czy warto uczyć się X - czy to przyszłościowe?"

        Okresy: '1m' (miesiąc), '3m' (kwartał), '6m' (6 miesięcy), '1y' (rok)
        """
        valid_periods = {'1m', '3m', '6m', '1y'}
        if period not in valid_periods:
            period = "3m"

        if technology:
            technology = sanitize_search_input(technology).lower()

        cache_params = {"period": period, "technology": technology}

        async def fetch():
            params = {"period": period}
            if technology:
                params["technology"] = technology
            return await api_client.get("/analytics/trends", params)

        result = await cache.get_or_set("market_trends", cache_params, fetch, TTL_TRENDS)
        data = result.get("data", {})

        return {
            "okres_analizy": period,
            "technologia": technology or "cały rynek IT",
            "ogólne_trendy": {
                "zmiana_liczby_ofert": data.get("job_count_change"),
                "zmiana_procentowa": data.get("job_count_change_pct"),
                "zmiana_wynagrodzeń": data.get("salary_change"),
                "dominujące_kontrakty": data.get("dominant_contract_type"),
            },
            "rosnące_technologie": data.get("growing_technologies", []),
            "malejące_technologie": data.get("declining_technologies", []),
            "nowe_wymagania": data.get("emerging_skills", []),
            "sezonowość": data.get("seasonality_note"),
            "prognoza_następne_3m": data.get("forecast_3m"),
        }

    @mcp.tool()
    async def get_required_skills(
        role: str,
        experience_level: Optional[str] = None
    ) -> dict:
        """
        Pobiera listę umiejętności najczęściej wymaganych dla danej roli.

        Użyj gdy użytkownik:
        - Chce wiedzieć czego się nauczyć żeby dostać pracę jako X
        - Pyta jakie skille są potrzebne dla konkretnej roli
        - Przygotowuje CV i chce wiedzieć co umieścić
        - Planuje ścieżkę nauki

        Dzieli umiejętności na: wymagane (must-have), mile widziane (nice-to-have),
        wyróżniające (differentiators).
        """
        role = sanitize_search_input(role)
        if experience_level:
            experience_level = experience_level.lower()

        cache_params = {"role": role, "level": experience_level}

        async def fetch():
            params = {"role": role}
            if experience_level:
                params["level"] = experience_level
            return await api_client.get("/analytics/required-skills", params)

        result = await cache.get_or_set("required_skills", cache_params, fetch, TTL_ANALYTICS)
        data = result.get("data", {})

        return {
            "rola": role,
            "poziom": experience_level or "wszystkie",
            "obowiązkowe_w_ponad_70_proc_ofert": data.get("must_have", []),
            "mile_widziane_30_70_proc": data.get("nice_to_have", []),
            "wyróżniające_poniżej_30_proc": data.get("differentiators", []),
            "certyfikaty_które_pomagają": data.get("helpful_certifications", []),
            "miękkie_umiejętności": data.get("soft_skills", []),
            "liczba_przeanalizowanych_ofert": data.get("analyzed_jobs_count"),
        }

    @mcp.tool()
    async def get_top_employers(
        technology: Optional[str] = None,
        city: Optional[str] = None,
        limit: int = 15
    ) -> dict:
        """
        Pobiera listę pracodawców z największą liczbą aktywnych ofert pracy IT.

        Użyj gdy użytkownik:
        - Chce wiedzieć kto dużo rekrutuje
        - Pyta o konkretnych pracodawców w danej technologii
        - Szuka firm rekrutujących w konkretnym mieście
        - Pyta "kto zatrudnia Python developerów w Krakowie?"
        """
        if technology:
            technology = sanitize_search_input(technology).lower()
        if city:
            city = sanitize_search_input(city)

        cache_params = {"technology": technology, "city": city, "limit": limit}

        async def fetch():
            params = {"limit": limit}
            if technology:
                params["technology"] = technology
            if city:
                params["city"] = city
            return await api_client.get("/analytics/top-employers", params)

        result = await cache.get_or_set("top_employers", cache_params, fetch, TTL_ANALYTICS)
        employers = result.get("employers", [])

        return {
            "filtr_technologia": technology or "wszystkie",
            "filtr_miasto": city or "wszystkie",
            "pracodawcy": [
                {
                    "firma": emp.get("company_name"),
                    "liczba_aktywnych_ofert": emp.get("active_jobs"),
                    "branża": emp.get("industry"),
                    "główna_lokalizacja": emp.get("main_location"),
                    "praca_zdalna": emp.get("offers_remote", False),
                    "technologie_szukane": emp.get("top_technologies", [])[:3],
                }
                for emp in employers
            ],
        }

Bezpieczeństwo - walidacja inputu i ochrona przed atakami

Serwer MCP przyjmuje dane od modelu AI - a model AI może być nakłoniony przez złośliwy prompt injection do przekazania niebezpiecznych danych. Walidacja jest kluczowa.

# security/validators.py
import re
import html
from typing import Optional

# Znaki dozwolone w wyszukiwaniach
SAFE_SEARCH_PATTERN = re.compile(r'^[a-zA-Z0-9ąćęłńóśźżĄĆĘŁŃÓŚŹŻ\s\-\+\#\.\/]+$')

# Słowa kluczowe które mogą wskazywać na atak
SUSPICIOUS_PATTERNS = [
    r'(drop|delete|truncate|alter|create|insert|update)\s+(table|database|schema)',
    r'union\s+select',
    r'

# Instalacja (jednorazowa)
npm install -g @modelcontextprotocol/inspector

# Uruchomienie inspektora z naszym serwerem
mcp-inspector python server.py

# Lub przez npx
npx @modelcontextprotocol/inspector python server.py

# tests/test_search.py
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from tools.search import format_job_for_ai
from security.validators import sanitize_search_input, validate_salary

# Testy walidacji - nie wymagają mock API

def test_sanitize_search_input_normal():
    """Normalne zapytanie powinno przejść bez zmian."""
    assert sanitize_search_input("Python developer") == "Python developer"

def test_sanitize_search_input_sql_injection():
    """SQL injection powinien być odrzucony."""
    with pytest.raises(ValueError, match="Niedozwolone wyrażenie"):
        sanitize_search_input("Python OR 1=1 UNION SELECT * FROM users")

def test_sanitize_search_input_too_long():
    """Za długi input powinien być przycięty."""
    long_input = "P" * 200
    result = sanitize_search_input(long_input)
    assert len(result) <= 100

def test_sanitize_search_path_traversal():
    """Path traversal powinien być odrzucony."""
    with pytest.raises(ValueError):
        sanitize_search_input("../../etc/passwd")

def test_validate_salary_valid():
    """Prawidłowe wynagrodzenie."""
    assert validate_salary(15000) == 15000

def test_validate_salary_too_low():
    """Za niskie wynagrodzenie."""
    with pytest.raises(ValueError, match="1000 PLN"):
        validate_salary(100)

def test_validate_salary_too_high():
    """Za wysokie wynagrodzenie."""
    with pytest.raises(ValueError, match="200 000 PLN"):
        validate_salary(500000)

def test_validate_salary_none():
    """None powinien być zwrócony bez zmian."""
    assert validate_salary(None) is None

def test_format_job_with_salary():
    """Formatowanie oferty z widełkami wynagrodzeń."""
    job = {
        "id": 1,
        "title": "Senior Python Developer",
        "company": "Tech Corp",
        "location": "Warszawa",
        "is_remote": True,
        "salary_min": 20000,
        "salary_max": 30000,
        "employment_type": "b2b",
        "experience_level": "senior",
        "technologies": ["Python", "Django", "PostgreSQL"],
        "created_at": "2026-06-21",
        "url": "https://2hr.pl/oferta/123/",
        "snippet": "Szukamy seniora...",
    }
    result = format_job_for_ai(job)
    assert result["tytuł"] == "Senior Python Developer"
    assert "20 000" in result["wynagrodzenie"]
    assert "B2B" in result["wynagrodzenie"]
    assert result["zdalna"] is True

def test_format_job_without_salary():
    """Formatowanie oferty bez widełek wynagrodzeń."""
    job = {
        "id": 2,
        "title": "Junior Developer",
        "company": "Startup",
        "location": "Kraków",
        "is_remote": False,
    }
    result = format_job_for_ai(job)
    assert result["wynagrodzenie"] == "nie podano"

# Testy asynchroniczne z mock API

@pytest.fixture
def mock_api_client():
    client = MagicMock()
    client.search_jobs = AsyncMock(return_value={
        "jobs": [
            {
                "id": 1,
                "title": "Python Developer",
                "company": "TestCorp",
                "location": "Warszawa",
                "is_remote": True,
                "salary_min": 15000,
                "salary_max": 25000,
                "employment_type": "b2b",
                "experience_level": "mid",
                "technologies": ["Python", "FastAPI"],
                "created_at": "2026-06-21",
                "url": "https://2hr.pl/oferta/1/",
                "snippet": "Szukamy mid Python developera"
            }
        ],
        "total": 1
    })
    return client

@pytest.fixture
def mock_cache():
    cache = MagicMock()
    # Cache zawsze zwraca None (cache miss) - wymusza wywołanie API
    cache.get_or_set = AsyncMock(side_effect=lambda prefix, params, fetch, ttl: fetch())
    return cache

@pytest.mark.asyncio
async def test_search_returns_formatted_results(mock_api_client, mock_cache):
    """Wyszukiwanie zwraca poprawnie sformatowane wyniki."""
    with patch('tools.search.api_client', mock_api_client), \
         patch('tools.search.cache', mock_cache):
        from tools.search import JobSearchInput

        params = JobSearchInput(keyword="Python", location="Warszawa")
        mock_api_client.search_jobs.assert_not_called()

@pytest.mark.asyncio
async def test_search_validates_keyword():
    """Wyszukiwanie odrzuca niebezpieczne słowa kluczowe."""
    from pydantic import ValidationError
    from tools.search import JobSearchInput

    with pytest.raises(ValidationError):
        JobSearchInput(keyword="DROP TABLE users --")

# macOS
~/Library/Application Support/Claude/claude_desktop_config.json

# Windows
%APPDATA%\Claude\claude_desktop_config.json

# Linux (Claude Desktop beta)
~/.config/claude/claude_desktop_config.json

{
  "mcpServers": {
    "2hr-jobs": {
      "command": "/path/to/venv-mcp/bin/python",
      "args": ["/path/to/jobs-mcp-server/server.py"],
      "env": {
        "JOBS_API_URL": "https://api.twoja-domena.pl/v1",
        "JOBS_API_KEY": "twoj_klucz_api",
        "REDIS_URL": "redis://localhost:6379/0"
      }
    }
  }
}

Znajdź zdalne oferty dla seniora Python z wynagrodzeniem powyżej 20 000 PLN B2B.

# Sprawdź logi Claude Desktop
# macOS
tail -f ~/Library/Logs/Claude/mcp*.log

# Windows
type %APPDATA%\Claude\logs\mcp.log

# Typowe problemy:
# 1. Zła ścieżka do Pythona - użyj pełnej ścieżki do venv
# 2. Brakujące zmienne środowiskowe - sprawdź sekcję "env" w konfiguracji
# 3. Błąd importu - uruchom server.py ręcznie i sprawdź błędy
# 4. Brakujące zależności - pip install -r requirements.txt w venv

# Lokalizacja pliku konfiguracyjnego Cursora
~/.cursor/mcp.json         # globalna konfiguracja
.cursor/mcp.json           # per-projekt (w katalogu projektu)

{
  "mcpServers": {
    "2hr-jobs": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "JOBS_API_URL": "https://api.twoja-domena.pl/v1",
        "JOBS_API_KEY": "twoj_klucz_api"
      }
    }
  }
}

# server_http.py - wersja z transportem HTTP/SSE
from mcp.server.fastmcp import FastMCP
from tools.search import register_search_tools
from tools.salary import register_salary_tools
from tools.analytics import register_analytics_tools
from resources.documentation import register_resources
from prompts.templates import register_prompts
import uvicorn
import os

mcp = FastMCP(
    name="2HR Jobs MCP Server",
    version="1.0.0",
)

# Rejestrujemy wszystkie narzędzia
register_search_tools(mcp)
register_salary_tools(mcp)
register_analytics_tools(mcp)
register_resources(mcp)
register_prompts(mcp)

# Eksportujemy aplikację ASGI dla HTTP transport
app = mcp.get_asgi_app()

if __name__ == "__main__":
    port = int(os.getenv("MCP_PORT", "8765"))
    host = os.getenv("MCP_HOST", "0.0.0.0")

    uvicorn.run(
        "server_http:app",
        host=host,
        port=port,
        reload=os.getenv("DEBUG", "false").lower() == "true",
        log_level=os.getenv("LOG_LEVEL", "info").lower(),
        workers=int(os.getenv("WORKERS", "1")),  # SSE wymaga sticky sessions
    )

{
  "mcpServers": {
    "2hr-jobs-remote": {
      "url": "https://mcp.twoja-domena.pl/sse",
      "transport": "sse"
    }
  }
}

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Instalujemy zależności systemowe (dla asyncpg i innych)
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Kopiujemy requirements i instalujemy
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Kopiujemy kod aplikacji
COPY . .

# Tworzymy nieprivilegowanego użytkownika
RUN useradd --create-home --shell /bin/bash mcp
USER mcp

# Port dla HTTP transport
EXPOSE 8765

# Healthcheck
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import httpx; httpx.get('http://localhost:8765/health').raise_for_status()"

# Uruchamiamy serwer HTTP
CMD ["python", "server_http.py"]

# docker-compose.yml
version: '3.9'

services:
  mcp-server:
    build: .
    restart: unless-stopped
    ports:
      - "8765:8765"
    environment:
      - JOBS_API_URL=${JOBS_API_URL}
      - JOBS_API_KEY=${JOBS_API_KEY}
      - REDIS_URL=redis://redis:6379/0
      - MCP_HOST=0.0.0.0
      - MCP_PORT=8765
      - LOG_LEVEL=info
      - WORKERS=1
    depends_on:
      redis:
        condition: service_healthy
    networks:
      - mcp-network

  redis:
    image: redis:7-alpine
    restart: unless-stopped
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    volumes:
      - redis_data:/data
    networks:
      - mcp-network

  nginx:
    image: nginx:alpine
    restart: unless-stopped
    ports:
      - "443:443"
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
      - /etc/letsencrypt:/etc/letsencrypt:ro
    depends_on:
      - mcp-server
    networks:
      - mcp-network

volumes:
  redis_data:

networks:
  mcp-network:
    driver: bridge

# nginx.conf - reverse proxy dla MCP
server {
    listen 443 ssl http2;
    server_name mcp.twoja-domena.pl;

    ssl_certificate /etc/letsencrypt/live/mcp.twoja-domena.pl/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mcp.twoja-domena.pl/privkey.pem;

    # SSE wymaga długich timeoutów
    proxy_read_timeout 300s;
    proxy_connect_timeout 10s;
    proxy_send_timeout 300s;

    # Wyłączamy buforowanie dla SSE
    proxy_buffering off;
    proxy_cache off;

    location / {
        proxy_pass http://mcp-server:8765;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        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;

        # Headers dla SSE
        proxy_set_header Cache-Control no-cache;
        add_header X-Accel-Buffering no;
    }

    location /health {
        proxy_pass http://mcp-server:8765/health;
        access_log off;
    }
}

server {
    listen 80;
    server_name mcp.twoja-domena.pl;
    return 301 https://$host$request_uri;
}

# Deployment
docker compose up -d

# Sprawdzenie statusu
docker compose ps
docker compose logs mcp-server

# Aktualizacja bez downtime
docker compose pull
docker compose up -d --no-deps mcp-server

# resources/documentation.py
from mcp.server.fastmcp import FastMCP

def register_resources(mcp: FastMCP) -> None:

    @mcp.resource("docs://api/overview")
    def get_api_overview() -> str:
        """
        Dokumentacja serwera MCP portalu 2hr.pl.
        Zawiera opis dostępnych narzędzi i ich parametrów.
        """
        return """
        # 2HR Jobs MCP Server - Dokumentacja

        ## Dostępne narzędzia

        ### Wyszukiwanie ofert
        - search_jobs - główne wyszukiwanie ofert pracy
        - find_remote_jobs - tylko oferty zdalne
        - find_jobs_in_city - oferty w konkretnym mieście
        - get_latest_jobs - najnowsze oferty

        ### Wynagrodzenia
        - get_salary_report - raport wynagrodzeń dla technologii
        - compare_salaries - porównanie dwóch technologii
        - check_salary_fairness - ocena uczciwości oferty

        ### Analityka rynkowa
        - get_top_technologies - ranking technologii
        - compare_roles - porównanie dwóch ról zawodowych
        - get_market_trends - trendy rynkowe
        - get_required_skills - wymagane umiejętności
        - get_top_employers - pracodawcy szukający pracowników

        ## Uwagi
        - Dane oparte na ofertach z portalu 2hr.pl
        - Wynagrodzenia w PLN brutto/miesiąc
        - Cache: oferty 5 min, wynagrodzenia 60 min, analityka 30 min
        """

    @mcp.resource("data://market/summary")
    async def get_market_summary() -> str:
        """Aktualny snapshot rynku pracy IT w Polsce."""
        from api.client import JobsApiClient
        client = JobsApiClient()
        try:
            data = await client.get("/analytics/market-summary")
            summary = data.get("data", {})
            return f"""
            Aktualny rynek pracy IT w Polsce ({summary.get('date', 'dziś')}):
            - Aktywnych ofert: {summary.get('active_jobs', 'N/A')}
            - Oferty zdalne: {summary.get('remote_percentage', 'N/A')}%
            - Top 3 technologie: {', '.join(summary.get('top_3_technologies', []))}
            - Mediana wynagrodzenia senior: {summary.get('senior_median_salary', 'N/A')} PLN
            """
        except Exception:
            return "Nie udało się pobrać aktualnych danych rynkowych."

# prompts/templates.py
from mcp.server.fastmcp import FastMCP

def register_prompts(mcp: FastMCP) -> None:

    @mcp.prompt()
    def job_search_assistant() -> str:
        """Asystent wyszukiwania pracy - punkt startowy dla sesji szukania pracy."""
        return """
        Jestem asystentem do szukania pracy IT w Polsce.
        Mam dostęp do aktualnych ofert pracy, danych o wynagrodzeniach
        i analityki rynku IT.

        Powiedz mi czego szukasz:
        - Jaką technologię lub rolę Cię interesuje?
        - W jakim mieście chcesz pracować (lub preferujesz zdalnie)?
        - Jaki poziom doświadczenia masz?
        - Jakie wynagrodzenie Cię interesuje?

        Mogę też pomóc ocenić czy konkretna oferta jest uczciwa rynkowo
        lub porównać wynagrodzenia dla różnych technologii.
        """

    @mcp.prompt()
    def salary_negotiation_assistant() -> str:
        """Asystent negocjacji wynagrodzenia."""
        return """
        Pomagam w negocjacjach wynagrodzenia w IT.
        Mam dostęp do aktualnych danych rynkowych z portalu 2hr.pl.

        Aby ocenić Twoją sytuację, potrzebuję:
        1. Jaką technologię/rolę wykonujesz?
        2. Ile lat doświadczenia masz?
        3. Jakie wynagrodzenie Ci zaproponowano?
        4. Czy to B2B (faktura) czy UoP (umowa o pracę)?
        5. W jakim mieście (lub zdalnie)?

        Na podstawie tych danych sprawdzę gdzie oferta stoi względem rynku
        i jak argumentować wyższe wynagrodzenie.
        """

# server.py - główny plik serwera
from mcp.server.fastmcp import FastMCP
from tools.search import register_search_tools
from tools.salary import register_salary_tools
from tools.analytics import register_analytics_tools
from resources.documentation import register_resources
from prompts.templates import register_prompts
import logging
import os

# Konfiguracja logowania
logging.basicConfig(
    level=getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper()),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def create_server() -> FastMCP:
    """Tworzy i konfiguruje serwer MCP."""
    mcp = FastMCP(
        name="2HR Jobs MCP Server",
        version="1.0.0",
    )

    # Rejestrujemy wszystkie komponenty
    register_search_tools(mcp)
    register_salary_tools(mcp)
    register_analytics_tools(mcp)
    register_resources(mcp)
    register_prompts(mcp)

    logger.info("Serwer MCP skonfigurowany - zarejestrowano narzędzia, zasoby i prompty")
    return mcp

# Dla stdio transport (Claude Desktop, Cursor)
if __name__ == "__main__":
    mcp = create_server()
    logger.info("Uruchamianie serwera MCP przez stdio...")
    mcp.run()

# monitoring/metrics.py
import time
import logging
from functools import wraps
from typing import Callable

logger = logging.getLogger(__name__)

# Prosta implementacja - w produkcji użyj Prometheus, Datadog lub podobnego

call_counts = {}
error_counts = {}
latency_sums = {}

def track_tool_call(tool_name: str):
    """Dekorator do śledzenia wywołań narzędzi."""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start_time = time.monotonic()
            call_counts[tool_name] = call_counts.get(tool_name, 0) + 1

            try:
                result = await func(*args, **kwargs)
                latency = time.monotonic() - start_time
                latency_sums[tool_name] = latency_sums.get(tool_name, 0) + latency

                if latency > 2.0:  # Wolne wywołanie - powyżej 2 sekund
                    logger.warning(f"Wolne wywołanie narzędzia {tool_name}: {latency:.2f}s")

                logger.debug(f"Tool {tool_name} completed in {latency:.3f}s")
                return result

            except Exception as e:
                error_counts[tool_name] = error_counts.get(tool_name, 0) + 1
                logger.error(f"Błąd w narzędziu {tool_name}: {e}", exc_info=True)
                raise

        return wrapper
    return decorator

def get_stats() -> dict:
    """Zwraca statystyki użycia narzędzi."""
    stats = {}
    for tool in call_counts:
        calls = call_counts[tool]
        errors = error_counts.get(tool, 0)
        avg_latency = latency_sums.get(tool, 0) / calls if calls > 0 else 0
        stats[tool] = {
            "calls": calls,
            "errors": errors,
            "error_rate": f"{(errors/calls*100):.1f}%" if calls > 0 else "0%",
            "avg_latency_ms": round(avg_latency * 1000, 1),
        }
    return stats

                    UŻYTKOWNICY
                        |
            +-----------+-----------+
            |                       |
    Claude Desktop              Cursor IDE
    (analiza rynku)          (dev szuka pracy)
            |                       |
            +----------+------------+
                       |
                  MCP Protocol
                       |
            +----------+------------+
            |                       |
     stdio transport           HTTP/SSE transport
     (lokalnie)                (produkcja)
            |                       |
            +----------+------------+
                       |
               MCP Server (Python)
               /api/client.py
                       |
               HTTP Request + API Key
                       |
            +----------+------------+
            |                       |
      REST API (PHP)            Redis Cache
      /v1/jobs/search           TTL 5-60 min
      /v1/salary/report
      /v1/analytics/
            |
            |
    +-------+--------+
    |                |
  MySQL 8.0       Elasticsearch
  (dane ofert)    (full-text search)

@mcp.tool()
async def generate_market_report(period: str = "1y") -> str:
    """
    Generuje szczegółowy raport rynku pracy IT.
    Może trwać kilka sekund - wyniki są streamowane.
    """
    # W FastMCP streaming jest transparentny
    # Duże stringi są automatycznie streamowane przez SSE transport
    report_parts = []

    report_parts.append("# Raport rynku pracy IT w Polsce\n\n")

    # Pobieramy dane sekcja po sekcji
    trends_data = await api_client.get_market_trends(period)
    report_parts.append(f"## Trendy\n{format_trends(trends_data)}\n\n")

    salary_data = await api_client.get_salary_overview()
    report_parts.append(f"## Wynagrodzenia\n{format_salaries(salary_data)}\n\n")

    tech_data = await api_client.get_top_technologies(limit=30)
    report_parts.append(f"## Technologie\n{format_technologies(tech_data)}\n\n")

    return "".join(report_parts)

# Zaawansowana autentykacja przez MCP headers
from mcp.server import Server
from mcp.types import ClientCapabilities

class AuthenticatedMCPServer:
    def __init__(self):
        self.server = Server("2HR Jobs Premium")

    async def get_user_context(self, request_headers: dict) -> dict:
        """Wyciąga i weryfikuje token użytkownika z headerów."""
        auth_header = request_headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return {"user_id": None, "plan": "free"}

        token = auth_header[7:]
        # Weryfikacja tokena przez API
        user_data = await api_client.verify_token(token)
        return user_data or {"user_id": None, "plan": "free"}

# Warm-up cache przy starcie serwera
async def warmup_cache():
    """Pre-ładuje najczęściej potrzebne dane do cache."""
    logger.info("Rozgrzewanie cache...")
    popular_technologies = ["Python", "JavaScript", "Java", "PHP", "Go", "React", "TypeScript"]

    tasks = [
        api_client.get_salary_report(tech)
        for tech in popular_technologies
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    success = sum(1 for r in results if not isinstance(r, Exception))
    logger.info(f"Cache rozgrzany - {success}/{len(popular_technologies)} technologii załadowanych")