AI: writing an MCP server for VictoriaLogs

#ai #python #todayileraned

In the previous article, we figured out what an MCP is in general, and created a basic MCP server that was connected to Windsurf — see AI: What is the MCP?

Now, let’s try to create something more useful, for example, an MCP server that will connect to VictoriaLogs and receive some data.

In fact, the VictoriaMetrics team is already making their own, so here we’ll “just play around” to see how MCP works on a more realistic example.

VictoriaLogs API

First, let’s try to make a request manually, and then wrap it in a Python code.

Open a local port for 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

To work with VictoriaLogs, we can use its HTTP API and make a request like this:

$ 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
...

Okay, everything is working.

Now let’s write a Python function that does the same thing.

We can even ask Cascade in Windsurf to do it for us:

However, we still had to run pip install -r requirements.txt ourselves in the console, because we do virtual env in Python (although later it did suggest running python -m venv venv).

As a result, we have the following code:

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")

In general, it’s fine. Simple enough, and it works:

$ 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
...

But then, when I asked Cascade to create an MCP server and add @tool, it fell into an infinite loop :-)

Okay, anyway — vibe-coding is not about us, we like to do everything with our own hands.

Creating an MCP server for VictoriaLogs

Let’s use the same FastMCP which we used in the previous post, and simplify the function a bit:

#!/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")

Install the fastmcp package directly:

$ pip install fastmcp

Try to start:

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

It seems to be working…

Let’s try to add to the Windsurf — edit the file ~/.codeium/windsurf/mcp\_config.json.

The my-mcp-server is the old one from the previous post, and now we're adding a new one - the victoria-logs-mcp:

{
  "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"
      ]
    }
  }
}

Run the Refresh in Windsurf Settings > MCP, and try to call our new server with the request like “find first log record with error”:

Wow :-)

And there is even an explanation of the error from the log -”I found a warning log from VictoriaMetrics about a failed target scrape. The system could not reach the blackbox exporter service, which is causing the issue.”

@tool: error analysis 5xx

First, let’s try it manually again with 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 ...
...

Now let’s add a new tool to our code:

...
@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}"
...

Save the changes, perform the Refresh — and a new analyze\_5xx\_logs tool has appeared for our MCP server: