Dashboards and reports remain the default interface for most forecasting systems, but they limit users to prespecified queries and views. In this post, we walk through the implementation of a conversational time series forecasting assistant built with ClickHouse, Amazon Bedrock, and LibreChat. This approach enables users to explore time series data interactively and adjust forecast parameters — such as quantile levels and prediction horizons — through natural language.
The solution is built around the Model Context Protocol (MCP), which defines a standard interface for connecting language models to external tools and services. Three MCP servers handle the core functionality: the official ClickHouse MCP server for data retrieval, a custom forecasting server wrapping Chronos - a time series foundation model developed by Amazon - on Amazon Bedrock, and a custom data visualization server for interactive Plotly charts. Claude Sonnet 4.6 on Amazon Bedrock orchestrates the tool calls.
The video below shows the assistant handling a retail sales forecasting scenario: exploring the available product sales time series, plotting a product's hourly sales over the most recent week, forecasting the next 48 hours, and repeating the analysis for additional products, with all results shown in a single combined chart.
Architecture
The three MCP servers are configured as Docker services in docker-compose.override.yml, alongside the LibreChat api service. The ClickHouse server uses the official mcp/clickhouse image; the forecasting and visualization servers are built from custom Docker images described below.
services:
api:
volumes:
- ./librechat.yaml:/app/librechat.yaml
mcp-clickhouse:
image: mcp/clickhouse
container_name: mcp-clickhouse
ports:
- 8001:8000
extra_hosts:
- "host.docker.internal:host-gateway"
environment:
- CLICKHOUSE_HOST=<CLICKHOUSE_HOST>
- CLICKHOUSE_USER=<CLICKHOUSE_USER>
- CLICKHOUSE_PASSWORD=<CLICKHOUSE_PASSWORD>
- CLICKHOUSE_MCP_SERVER_TRANSPORT=sse
- CLICKHOUSE_MCP_BIND_HOST=0.0.0.0
chronos-forecasting:
build:
context: ./chronos-forecasting
ports:
- "8002:8002"
environment:
- AWS_DEFAULT_REGION=eu-west-1
- AWS_ACCESS_KEY_ID=<AWS_ACCESS_KEY_ID>
- AWS_SECRET_ACCESS_KEY=<AWS_SECRET_ACCESS_KEY>
data-visualization:
build:
context: ./data-visualization
ports:
- "8003:8003"
- "8004:8004"
volumes:
- ./plots:/plots
Note that the visualization server exposes two ports: 8003 for the MCP server (SSE transport) and 8004 for the HTTP file server that serves the generated HTML charts. The /plots directory is mounted as a volume so charts persist across container restarts.
The servers are registered in librechat.yaml using SSE transport:
mcpServers:
clickhouse-playground:
type: sse
url: http://host.docker.internal:8001/sse
chronos-forecasting:
type: sse
url: http://host.docker.internal:8002/sse
data-visualization:
type: sse
url: http://host.docker.internal:8003/sse
The LibreChat modelSpecs object ties the model, the MCP servers, and the system prompt together into a named assistant configuration:
modelSpecs:
list:
- name: "forecasting-assistant"
label: "Forecasting Assistant"
default: true
artifacts: true
mcpServers:
- "clickhouse-playground"
- "chronos-forecasting"
- "data-visualization"
preset:
endpoint: "bedrock"
model: "eu.anthropic.claude-sonnet-4-6"
region: "eu-west-1"
promptPrefix: |
You are a time series forecasting assistant.
Do not alter the time frequency of the data unless the user explicitly requests it.
To visualize data, always use the `visualize_data` tool. Never write code to generate plots.
When visualizing multiple series, always pass them together to the `visualize_data` tool in a single call,
as it generates subplots automatically.
Always render the URL returned by the `visualize_data` tool as an artifact using this exact format:
:::artifact{identifier="<IDENTIFIER>" type="text/html" title="<TITLE>"}
\`\`\`
<!DOCTYPE html>
<html>
<body>
<iframe src="<URL>" width="100%" height="800px" frameborder="0"></iframe>
</body>
</html>
\`\`\`
:::
Never return the raw URL and never ask the user to open the URL themselves.
The system prompt instructs the model to wrap chart URLs in LibreChat Artifacts format, so the LibreChat UI renders the Plotly chart directly rather than displaying a raw localhost URL.
Forecasting MCP Server
The forecasting server is built with FastMCP and exposes a single generate_forecasts tool. It invokes the Chronos endpoint deployed on Amazon Bedrock via boto3, passing the historical values as context along with the desired prediction length and quantile levels.
Since Chronos performs zero-shot forecasting without fine-tuning, the server works out of the box on any time series. For instructions on deploying Chronos on Amazon Bedrock, see this previous post.
server.py
import json
import boto3
from mcp.server.fastmcp import FastMCP
# Create the FastMCP server
mcp = FastMCP(
name="chronos-forecasting",
host="0.0.0.0",
port=8002
)
# Register the tool with the FastMCP server
@mcp.tool()
def generate_forecasts(
target: list[float],
prediction_length: int,
quantile_levels: list[float]
) -> dict:
"""
Generate probabilistic time series forecasts using Chronos on Amazon Bedrock.
Parameters:
===============================================================================
target: list of float.
The historical time series values used as context.
prediction_length: int.
The number of future time steps to predict.
quantile_levels: list of float.
The quantiles to be predicted at each future time step.
Returns:
===============================================================================
dict
Dictionary with predicted mean and quantiles at each future time step.
"""
# Create the Bedrock client
bedrock_runtime_client = boto3.client("bedrock-runtime")
# Invoke the Bedrock endpoint
response = bedrock_runtime_client.invoke_model(
modelId="<bedrock-endpoint-arn>",
body=json.dumps({
"inputs": [{
"target": target
}],
"parameters": {
"prediction_length": prediction_length,
"quantile_levels": quantile_levels
}
})
)
# Extract the forecasts
forecasts = json.loads(response["body"].read()).get("predictions")[0]
# Return the forecasts
return forecasts
# Run the FastMCP server with SSE transport
if __name__ == "__main__":
mcp.run(transport="sse")
The server's requirements.txt and Dockerfile are reported below:
requirements.txt
mcp
boto3
Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY server.py .
CMD ["python", "server.py"]
Data Visualization MCP Server
The data visualization server is also built with FastMCP. It exposes a visualize_data tool that takes historical data and optional forecast outputs, builds a Plotly figure with one subplot per time series, saves it as a self-contained HTML file, and returns the URL. A lightweight HTTP server running in a background thread serves the /plots directory on port 8004.
Using a dedicated MCP server for data visualization — rather than having the model generate chart code directly — ensures consistent styling and reproducible outputs across sessions, which matters when the assistant is used for reporting.
The tool input schema is designed to accept the raw output of ClickHouse queries directly, so the model does not need to reformat data between tool calls:
server.py
import os
import uuid
import threading
import http.server
import functools
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from mcp.server.fastmcp import FastMCP
# Create the FastMCP server
mcp = FastMCP(
name="data-visualization",
host="0.0.0.0",
port=8003
)
# Register the tool with the FastMCP server
@mcp.tool()
def visualize_data(
inputs: dict
) -> str:
"""
Plot one or more time series with optional forecasts and return
the URL of the interactive HTML chart.
Parameters
===============================================================================
inputs : dict
A dictionary with the following keys:
"data" (required) : dict
A dictionary where each key is a series name and each value is the
raw output of a ClickHouse query, with the following fields:
- "columns" : list of strings, must contain "timestamp" and one value column
- "rows" : list of [timestamp_str, float] pairs
Example:
{
"series_1": {
"columns": ["timestamp", "<VALUE>"],
"rows": [
["2026-01-01", 1.0],
["2026-01-02", 2.0]
]
},
"series_2": {
"columns": ["timestamp", "<VALUE>"],
"rows": [
["2026-01-03", 3.0],
["2026-01-04", 4.0]
]
},
}
"forecasts" (optional) : dict
Forecasts for the same time series in "data". Each key is a series
name matching a key in "data", and each value is a dictionary with
the following fields:
- "timestamp" : list of strings representing datetimes
- "mean" : list of floats (mean forecast)
- "<quantile>" : list of floats for each quantile level, e.g.
"0.05" and "0.95" for a 90% prediction interval.
Example:
{
"series_1": {
"timestamp": ["2026-01-01", "2026-01-02"],
"mean": [1.0, 2.0],
"0.1": [0.5, 1.5],
"0.5": [1.0, 2.0],
"0.9": [1.5, 2.5],
},
"series_2": {
"timestamp": ["2026-01-03", "2026-01-04"],
"mean": [3.0, 4.0],
"0.1": [2.5, 3.5],
"0.5": [3.0, 4.0],
"0.9": [3.5, 4.5],
},
}
Returns
===============================================================================
str
The URL of the interactive HTML chart.
"""
# Extract the data and forecasts
data = inputs["data"]
forecasts = inputs.get("forecasts", {})
# Parse the data
parsed_data = {}
for series, query_result in data.items():
value_col = [c for c in query_result["columns"] if c != "timestamp"][0]
idx = query_result["columns"].index
parsed_data[series] = {
"timestamp": [row[idx("timestamp")] for row in query_result["rows"]],
"values": [row[idx(value_col)] for row in query_result["rows"]]
}
# Create the figure
fig = make_subplots(
rows=len(data),
subplot_titles=list(data.keys())
)
# Update the figure layout
fig.update_layout(
height=250 * len(data),
paper_bgcolor="white",
plot_bgcolor="white",
margin=dict(t=50, b=50, l=50, r=50),
hovermode="x unified",
hoverlabel=dict(
namelength=-1
),
legend=dict(
font=dict(
color="#24292f",
size=12
),
)
)
# Update the subplots titles
fig.update_annotations(
font=dict(
color="#24292f",
size=14
),
)
# Generate the subplots
for i, series in enumerate(data):
# Plot the forecasts
if series in forecasts:
# Extract the predicted quantiles
q = sorted([float(k) for k in forecasts[series] if k not in ("mean", "timestamp")])
# Extract the lower and upper bound of the prediction interval
q_min, q_max = q[0], q[-1]
# Plot the upper bound of the prediction interval
fig.add_trace(
go.Scatter(
x=forecasts[series]["timestamp"],
y=forecasts[series][str(q_max)],
name=f"Predicted Q{q_max:,.1%}",
hovertemplate="%{fullData.name}: %{y:,.0f}<extra></extra>",
showlegend=False,
mode="lines",
line=dict(
width=0.5,
color="#c2e5ff",
),
),
row=i + 1,
col=1
)
# Plot the lower bound of the prediction interval
fig.add_trace(
go.Scatter(
x=forecasts[series]["timestamp"],
y=forecasts[series][str(q_min)],
name=f"Predicted Q{q_min:,.1%}",
hovertemplate="%{fullData.name}: %{y:,.0f}<extra></extra>",
showlegend=False,
mode="lines",
line=dict(
width=0.5,
color="#c2e5ff",
),
fillcolor="#c2e5ff",
fill="tonexty",
),
row=i + 1,
col=1
)
# Plot the predicted median if available, otherwise fall back to the predicted mean
fig.add_trace(
go.Scatter(
x=forecasts[series]["timestamp"],
y=forecasts[series]["0.5" if 0.5 in q else "mean"],
name=f"Predicted {'Median' if 0.5 in q else 'Mean'}",
hovertemplate="%{fullData.name}: %{y:,.0f}<extra></extra>",
showlegend=i == 0,
mode="lines",
line=dict(
color="#0588f0",
width=1,
dash="dot"
)
),
row=i + 1,
col=1
)
# Plot the data
fig.add_trace(
go.Scatter(
x=parsed_data[series]["timestamp"],
y=parsed_data[series]["values"],
name="Historical Data",
hovertemplate="%{fullData.name}: %{y:,.0f}<extra></extra>",
mode="lines",
showlegend=i == 0,
line=dict(
color="#838383",
width=1
)
),
row=i + 1,
col=1
)
# Update the subplot's x-axis
fig.update_xaxes(
type="date",
tickformat="%b %d %Y<br>(%a) %H:%M",
tickangle=0,
mirror=True,
linecolor="#cecece",
gridcolor="#e8e8e8",
gridwidth=0.5,
tickfont=dict(
color="#24292f",
size=10
),
row=i + 1,
col=1
)
# Update the subplot's y-axis
fig.update_yaxes(
tickformat=",.0f",
mirror=True,
linecolor="#cecece",
gridcolor="#e8e8e8",
gridwidth=0.5,
tickfont=dict(
color="#24292f",
size=10
),
row=i + 1,
col=1
)
# Save the figure to an HTML file
os.makedirs("/plots", exist_ok=True)
filename = f"plot_{uuid.uuid4().hex}.html"
fig.write_html(f"/plots/{filename}", full_html=True, include_plotlyjs="cdn")
# Return the URL of the HTML file
return f"http://localhost:8004/{filename}"
if __name__ == "__main__":
# Serve the /plots directory over HTTP on port 8004
def _serve_plots():
handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory="/plots")
with http.server.HTTPServer(("0.0.0.0", 8004), handler) as httpd:
httpd.serve_forever()
# Start the HTTP server
threading.Thread(target=_serve_plots, daemon=True).start()
# Run the FastMCP server with SSE transport
mcp.run(transport="sse")
The requirements.txt and Dockerfile for this server follow the same pattern as the forecasting server:
requirements.txt
mcp
plotly
Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY server.py .
CMD ["python", "server.py"]
With ClickHouse for data, Chronos on Amazon Bedrock for zero-shot forecasts, and LibreChat orchestrated via MCP, the system provides an end-to-end conversational interface for querying, forecasting, and visualizing multiple time series interactively. The full post covers the complete LibreChat configuration and walks through a sample conversation with the assistant.
Top comments (0)