Modele językowe w wersji podstawowej, nie są w stanie zrobić nic, oprócz przetwarzania informacji. W dzisiejszych czasach, jednak wymagamy aby systemy potrafiły same reagować na zdarzenia i same sięgać po informacje z różnych źródeł. Do tego celu powstał protokół MCP (więcej informacji pod tym linkiem).
W dzisiejszym praktycznym przykładzie skupimy się na utworzeniu usługi dla administratorów, która będzie w stanie sprawdzić (gdy zapytamy ją o to w języku naturalnym), czy dany komputer jest w naszej sieci. Przykład może jest trywialny, ale uważam że na takich najlepiej się uczyć rozumieć trudne zagadnienia.
Środowisko wirtualne
Zacznijmy od podstaw. Utwórzmy nasze odizolowane środowisko i zainstalujmy zależności:python -m uv init
uv add mcp pythonping fastmcp
.venv\Scripts\activate
Import bibliotek
Jedną z pierwszych trudności które napotkamy, jest wybór odpowiedniej wersji. Biblioteki FastMCP w wersjach 1.0 i 2.0 są między sobą niekompatybilne już na poziomie definiowania importu.
Dla wersji FastMCP 1.0 użyjemy:
from mcp.server import FastMCP
Zaś dla wersji FastMCP 2.0 skorzystamy z (ją wybieramy w poniższym przykładzie):
from fastmcp import FastMCP
Definicja loggera
Każdy nasz większy projekt, powinniśmy zacząć od zdefiniowania loggera. Pozwoli nam na zbieranie pełnych informacji, co się dzieje w danym momencie w aplikacji.
import logging
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s: %(asctime)s \t%(message)s",
datefmt="%H:%M:%S"
)
logger = logging.getLogger(__name__)
Narzędzia
Przejdźmy teraz do właściwego kodu. Utwórzmy naszą funkcję, która sprawdzi czy dany serwer jest dostępny w sieci - czyli czy odpowiada na ping.
from pythonping import ping as py_ping
@mcp.tool()
def ping(host: str) -> str:
logger.info(f'"PING {host}"')
try:
response = py_ping(host, count=1, timeout=2)
if response.success():
msg = f'"PING {host}" 200 OK (czas: {response.rtt_avg_ms:.1f} ms)'
logger.debug(msg)
return "online"
else:
msg = f'"PING {host}" 408 TIMEOUT'
logger.warning(msg)
return "offline"
except Exception as e:
msg = f'"PING {host}" 500 ERROR ({e})'
logger.error(msg)
return "offline"
Rejestracja narzędzi
Każde zdefiniowane przez nas narzędzie, należy podpiąć pod naszą instancję obiektu klasy "FastMCP". Pozwoli ona na późniejsze wywołanie tych metod po przez API.
def register_tools(mcp):
@mcp.tool()
def ping(host: str) -> str:
pass
mcp = FastMCP()
register_tools(mcp)
Testowanie serwera
Zanim przejdziemy do wywołania naszych akcji, należy uruchomić inspektor. Dzięki niemu będziemy mogli podejrzeć i w łatwy sposób przetestować wszystkie akcje. Należy w konsoli uruchomić:
npx @modelcontextprotocol/inspector
i otworzyć w przeglądarce adres url:
Open inspector with token pre-filled:
http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=f5544537f23039d2119aaa0a92f6610ed27c1a99cc5758e5e19de8d590ae16bd
Następnie w zależności od preferowanej metody połączenia, dobieramy odpowiednie opcje:
1. Stdio - komunikacja przez standardowe operacje wejścia/wyjścia
W niej, program uruchamiamy po przez:
mcp.run(transport="stdio")
2. Streamable - komunikacja przez protokół http
W nim, program uruchamiamy po przez metodę:
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")
Wymaga ustawienia w inspektorze:
Teraz pozostało tylko uruchomić nasz plik z serwerem:
python server.py
I kliknąć connect w inspektorze.
*Po poprawnym połączeniu, powinna się zapalić zielona kontrolka oraz pojawić się nam panel zarządzania.
Testy
Z poziomu samego "MCP inspektora", możemy wywołać nasze narzędzia ręcznie. Dzięki temu mamy możliwość zasymulowania bezpośredniego wykonania akcji, bez wywołania jej pośrednio przez LLM.
Dla testu, możesz w zakładce tools, spróbować wykonać polecenie "ping" do innego urządzenia.
Po rozwinięciu zakładki "historia", mamy dostęp do wyników naszych poleceń.
Pełny kod:
import logging
from fastmcp import FastMCP
from pythonping import ping as py_ping
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s: %(asctime)s \t%(message)s",
datefmt="%H:%M:%S"
)
logger = logging.getLogger(__name__)
def register_tools(mcp: FastMCP) -> None:
"""
Registers available tools within the given MCP server instance.
Args:
mcp (FastMCP): The FastMCP server instance to which the tools will be added.
Returns:
None
"""
@mcp.tool()
def ping(host: str) -> str:
"""
Sends an ICMP echo (ping) request to the specified host and returns its network status.
Args:
host (str): The target hostname or IP address to check (e.g., "8.8.8.8" or "example.com").
Returns:
str: One of the following status values:
- `"online"` – if the host responded successfully to the ping request
- `"offline"` – if the host did not respond or a timeout/error occurred
"""
logger.info(f'"PING {host}"')
try:
response = py_ping(host, count=1, timeout=2)
if response.success():
msg = f'"PING {host}" 200 OK (czas: {response.rtt_avg_ms:.1f} ms)'
logger.debug(msg)
return "online"
else:
msg = f'"PING {host}" 408 TIMEOUT'
logger.warning(msg)
return "offline"
except Exception as e:
msg = f'"PING {host}" 500 ERROR ({e})'
logger.error(msg)
return "offline"
def main() -> None:
"""
Initializes the MCP server, registers tools, and starts the HTTP service.
"""
server = FastMCP()
register_tools(server)
logger.info("Starting MCP Ping Server on http://127.0.0.1:8000/mcp")
server.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")
if __name__ == "__main__":
main()
W części drugiej, zajmiemy się naszym klientem. Będzie się odwoływał do naszego serwera, aby skorzystać z jego narzędzi.
Komentarze