В попередньому матеріалі розібрались з тим, що таке MCP взагалі, і створили дуже простенький сервер, який підключили до Windsurf – див. AI: що таке той MCP?

Тепер – давайте спробуємо створити щось більш корисне, наприклад – MCP-сервер, який буде підключатись до VictoriaLogs та отримувати якісь дані.

Насправді команда VictoriaMetrcis вже робить власний, тому тут ми “просто пограємось”, аби подивитись на MCP на більш реальному прикладі.

Зміст

VictoriaLogs API

Спершу спробуємо зробити запит руками, а потім “загорнемо” його в код на Python.

Відкриваємо локальний порт на VictoriaLogs:

$ kubectl port-forward svc/atlas-victoriametrics-victoria-logs-single-server 9428
Forwarding from 127.0.0.1:9428 -> 9428
Forwarding from [::1]:9428 -> 9428

Для роботи з VictoriaLogs ми можемо використати її HTTP API і зробити запит на кшталт такого:

$ curl -X POST http://localhost:9428/select/logsql/query -d 'query=error' -d 'limit=1'
{"_time":"2025-05-08T11:13:44.36173829Z","_stream_id":"0000000000000000ae9096a01bcfdf58cd1159fc206f3aea","_stream":"{namespace=\"ops-monitoring-ns\"}","_msg":"2025-05-08T11:13:44.361Z\twarn\tVictoriaMetrics/lib/promscrape/scrapework.go:383\tcannot scrape target \"http://atlas-victoriametrics-prometheus-blackbox-exporter.ops-monitoring-ns.svc:9115
...

Окей, ту все працює.

Тепер давайте напишемо Python-функцію, яка зробить те саме.

Можемо навіть попросити Cascade у Windsurf це зробити за нас:

Правда, pip install -r requirements.txt все ж довелось виконувати самому в консолі, бо робимо в Python virtual env (хоча пізніше він все ж запропонував виконати python -m venv venv).

В результаті маємо такий код:

import requests

def query_victoria_logs(query: str, limit: int = 1, host: str = "localhost", port: int = 9428) -> dict:
    """
    Query VictoriaLogs endpoint using the logsql/query endpoint.
    
    Args:
        query: The search query to execute
        limit: Maximum number of results to return
        host: VictoriaLogs host (default: localhost)
        port: VictoriaLogs port (default: 9428)
        
    Returns:
        Dictionary containing the response from VictoriaLogs
    """
    url = f"http://{host}:{port}/select/logsql/query"
    params = {
        'query': query,
        'limit': limit
    }
    
    try:
        response = requests.post(url, data=params)
        response.raise_for_status()  # Raise an exception for bad status codes
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error querying VictoriaLogs: {e}")
        return None

if __name__ == "__main__":
    # Example usage
    result = query_victoria_logs(query="error", limit=1)
    if result:
        print("Query results:")
        print(result)
    else:
        print("No results or error occurred")

В цілому – нормально. Достатньо просто, і працює:

$ python victoria_logs_client.py 
Query results:
{'_time': '2025-05-10T11:43:43.694956594Z', '_stream_id': '0000000000000000ae9096a01bcfdf58cd1159fc206f3aea', '_stream': '{namespace="ops-monitoring-ns"}', '_msg': '2025-05-10T11:43:43.694Z\twarn\tVictoriaMetrics/lib/promscrape/scrapework.go:383\tcannot scrape target "http://atlas-victoriametrics-prometheus-blackbox-exporter.ops-monitoring-ns.svc:9115/probe
...

Але далі, коли я попросив Cascade створити MCP-сервер і додати @tool – то він трохи впав в безкінечний цикл 🙂

Окей, anyway – vibe-кодінг це не про нас, ми любимо все робити власними руками.

Створення MCP-серверу для VictoriaLogs

Використовуємо той самий FastMCP, і трохи спростимо функцію:

#!/usr/bin/env python3

from fastmcp import FastMCP
import requests

mcp = FastMCP("VictoriaLogs MCP Server")

@mcp.tool()
def query_victorialogs(query: str, limit: int = 1) -> str:
    """
    Run a LogsQL query against VictoriaLogs and return raw result.
    """
    url = "http://localhost:9428/select/logsql/query"
    data = {
        "query": query,
        "limit": str(limit)
    }

    try:
        response = requests.post(url, data=data)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        return f"Error: {e}"

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

Встановлюємо fastmcp пакет напряму:

$ pip install fastmcp

Пробуємо запустити:

$ ./victoria_logs_client.py 
[05/10/25 14:56:43] INFO     Starting server "VictoriaLogs MCP Server"...

Наче навіть працює…

Пробуємо додати до Windsufr – my-mcp-server це старий, з попереднього поста, додаємо новий – victoria-logs-mcp.

Редагуємо файл ~/.codeium/windsurf/mcp_config.json:

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "/home/setevoy/Scripts/Python/MCP/my-mcp-server/.venv/bin/mcp",
      "args": [
        "run",
        "/home/setevoy/Scripts/Python/MCP/my-mcp-server/mcp_server.py"
      ]
    },
    "victoria-logs-mcp": {
      "command": "/home/setevoy/Scripts/Python/MCP/my-mcp-server/.venv/bin/python",
      "args": [
        "/home/setevoy/Scripts/Python/MCP/my-mcp-server/victoria_logs_client.py"
      ]
    }
  }
}

Робимо Refresh у Windsurf Settings > MCP, і спробуємо викликати наш новий сервер з запитом “find first log record woth error“:

Офігєть 🙂

І навіть є пояснення помилки з логу – “I found a warning log from VictoriaMetrics about a failed target scrape. The system couldn’t reach the blackbox exporter service, which is causing the issue“.

@tool: аналіз помилок 5хх

Спочатку давайте знову спробуємо з curl:

$ curl -X POST http://localhost:9428/select/logsql/query -d 'query=_msg:~"status=50[024]"' -d 'limit=100' -d 'format=json'
...
{"_time":"2025-05-09T13:59:40.41620395Z","_stream_id":"0000000000000000ae9096a01bcfdf58cd1159fc206f3aea","_stream":"{namespace=\"ops-monitoring-ns\"}","_msg":"level=warn ...
...

Тепер додаємо новий tool в наш код:

...
@mcp.tool()
def analyze_5xx_logs(limit: int = 100) -> str:
    """
    Analyze recent logs with 5xx HTTP status codes by parsing NDJSON response from VictoriaLogs.
    """
    url = "http://localhost:9428/select/logsql/query"
    query = '_msg:~"status=50[0234]"'

    data = {
        "query": query,
        "limit": str(limit),
        "format": "json"
    }

    try:
        response = requests.post(url, data=data)
        response.raise_for_status()

        # NDJSON: parse each line as a separate JSON object
        entries = []
        for line in response.text.strip().splitlines():
            try:
                entry = json.loads(line)
                entries.append(entry)
            except json.JSONDecodeError:
                continue

    except requests.RequestException as e:
        return f"Request error: {e}"

    if not entries:
        return f"No logs found matching: {query}"

    messages = [entry.get("_msg", "") for entry in entries]
    combined = "\n".join(messages[:limit])
    return f"Found {len(messages)} logs matching `{query}`:\n\n{combined}"
...

Зберігаємо зміни в коді, робимо Reload – і для нашого MCP-серверу з’явився новий tool analyze_5xx_logs: