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 sięgać po informacje z źródeł, w miarę potrzeb. 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 (wybieramy ją w naszym 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 o naszej aplikacji. O tym co się dzieje w aktualnym momencie, jak i wydarzyło się w przeszłości.
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ą główną instancję obiektu klasy "FastMCP". Pozwoli ona po zarejestrowaniu, 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 testowania naszych akcji, należy uruchomić inspektor. Dzięki niemu będziemy mogli podejrzeć i w łatwy sposób przetestować wszystkie nasze funkcje. 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.
Testów ciąg dalszy...
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ń.
Gdy już wszystko działa
Czas na uruchomienie naszej usługi na serwerze. Poniżej przedstawiam kompletny kod który należy zapisać w pliku mcp_server.py.
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()
Sam skrypt uruchamiamy po przez:
python mcp_server.py
Po wszystkim powinniśmy otrzymać:
[12/10/25 17:41:27] INFO Starting MCP server 'FastMCP-fd70' with transport 'http' on http://0.0.0.0:8000/mcp server.py:2055
INFO: Started server process [24016]
INFO: Waiting for application startup.
INFO: 17:41:27 StreamableHTTP session manager started
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Czas na testowego klienta
Z racji że udało się nam poprawnie uruchomić naszą instancje serwera MCP, możemy go szybko przetestować poniższym prostym skryptem:
import asyncio
from fastmcp import Client, FastMCP
server = FastMCP()
client = Client("http://127.0.0.1:8000/mcp")
async def main():
async with client:
# Check if server is alive
await client.ping()
# List available operations from server
tools = await client.list_tools()
# resources = await client.list_resources()
# prompts = await client.list_prompts()
# Execute operation from server
result = await client.call_tool("ping", {"host": "127.0.0.1"})
print(result)
asyncio.run(main())
Po uruchomieniu powinniśmy otrzymać:
CallToolResult(content=[TextContent(type='text', text='127.0.0.1 - online', annotations=None, meta=None)], structured_content={'result': '127.0.0.1 - online'}, meta=None, data='127.0.0.1 - online', is_error=False)
W następnej części zajmiemy się stworzeniem w pełni autonomicznego systemu. Pełny kod przykładu znajdziesz tutaj.
Komentarze