How to Build a Personal Financial Assistant With MCP, Python, and Deterministic Metrics

LLMs are great at writing market commentary. The problem is they can sound confident even when they have not looked at any data. That’s fine for casual chat. It’s not fine if you’re building a feature for a product, an internal tool, or anything a user might rely on. In this guide, we’ll build a small financial assistant that answers by fetching real data through MCP (Model Context Protocol), then computing the numbers in Python. The LLM’s job is only to narrate the computed facts. It does not invent metrics, and it does not do the math. By the end, you’ll have two outputs you can actually plug into a product flow. A single-ticker market brief, and a watchlist snapshot that compares multiple tickers on volatility and drawdown, with the tool calls traced so you can see exactly what data was used. What is MCP? MCP (Model Context Protocol) is basically a standard way for an LLM app to discover and call external tools. Instead of hardcoding a bunch of function schemas or building custom connectors per framework, you plug into an MCP server and the tools become “available” in a consistent format. For product teams, this matters because it reduces integration churn. Tool discovery is predictable, you’re not rewriting wrappers every time your stack changes, and you get a clean separation between the model and the data layer. In our case, that data layer is EODHD’s MCP server, which exposes market data tools the assistant can call whenever it needs prices or fundamentals. EODHD’s MCP server One important clarification for this tutorial. We’re using MCP purely as the data access layer. The model does not decide which tools to call or what parameters to pass. We do that deterministically in Python, then hand the model a facts object and let it write the narrative. This keeps the output grounded and makes the system much easier to trust and debug. Architecture: The “narrator” pattern Here’s the architecture we’re using in this guide. The idea is simple. We separate “getting facts” from “writing words”. The model only does the second part. First, the user asks a question like “Give me a 30-day brief for AAPL” or “Compare TSLA, NVDA, AMZN over the last 60 days”. That raw text goes into a tiny parser. The parser is intentionally boring. It only extracts what the system needs to operate. A list of tickers and a lookback window. Once we have tickers and dates, we fetch data through MCP. In this case, MCP is connected to EODHD’s MCP server. So instead of the assistant guessing prices or fundamentals, it calls tools like “get historical prices” and “get fundamentals”. At this point we have raw data. Nothing has been computed yet, and the model has not written a single sentence. Then Python takes over. This is where we compute everything deterministic. Returns, volatility, max drawdown, trend slope, and a simple volatility regime label. For watchlists, we align returns and compute correlation. These numbers are the backbone of the output. If you rerun the same query with the same window, you should get the same metrics. Only after that do we involve the LLM. We pass it a compact facts object. It contains the metrics we computed, plus a few clean fundamentals fields. The prompt is strict. Use only these facts. No extra numbers. No guessing. The model’s job is to turn the facts into a clean note that feels like something a product would show. Finally, the assistant returns a structured response object. Not just text. You get: answer (the narrative) metrics (the exact computed numbers) data_used (tickers, date range, and which tools were called) tool_trace_id (a trace id you can log, debug, or attach to monitoring) answer (the narrative) answer metrics (the exact computed numbers) metrics data_used (tickers, date range, and which tools were called) data_used tool_trace_id (a trace id you can log, debug, or attach to monitoring) tool_trace_id This pattern is B2B-friendly for a very practical reason. It reduces hallucinations because the model isn’t doing analysis. It makes numbers repeatable because Python computes them. And it’s easy to audit because you can always show what data was fetched, what window was used, and which tool calls happened. Step 1. MCP client wrapper (client.py) client.py Before we touch any “assistant logic”, we need one thing. A tiny MCP client wrapper that can connect to EODHD’s MCP server and call tools reliably. That’s it. This file does three jobs: opens a streamable HTTP MCP session calls a tool with a timeout and a small retry loop returns the tool output plus a small metadata object we can later attach to logs and traces opens a streamable HTTP MCP session calls a tool with a timeout and a small retry loop returns the tool output plus a small metadata object we can later attach to logs and traces Here’s the complete client.py. client.py import time import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client class EODHDMCP: def __init__(self, apikey, base_url=None): self.apikey = apikey self.base_url = base_url or "https://mcp.eodhd.dev/mcp" self._tools = None def _url(self): return f"{self.base_url}?apikey={self.apikey}" def _open(self): return streamablehttp_client(self._url()) async def list_tools(self): if self._tools is not None: return self._tools async with self._open() as (read, write, _): async with ClientSession(read, write) as s: await s.initialize() resp = await s.list_tools() self._tools = [t.name for t in resp.tools] return self._tools async def call_tool(self, name, args, trace_id, timeout_s=25, retries=1): last = None for attempt in range(retries + 1): t0 = time.time() try: async with self._open() as (read, write, _): async with ClientSession(read, write) as s: await s.initialize() out = await asyncio.wait_for(s.call_tool(name, args), timeout=timeout_s) dt = time.time() - t0 meta = {"trace_id": trace_id, "tool": name, "args": args, "latency_s": round(dt, 3)} return out, meta except Exception as e: last = e if attempt < retries: await asyncio.sleep(0.25) raise last import time import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client class EODHDMCP: def __init__(self, apikey, base_url=None): self.apikey = apikey self.base_url = base_url or "https://mcp.eodhd.dev/mcp" self._tools = None def _url(self): return f"{self.base_url}?apikey={self.apikey}" def _open(self): return streamablehttp_client(self._url()) async def list_tools(self): if self._tools is not None: return self._tools async with self._open() as (read, write, _): async with ClientSession(read, write) as s: await s.initialize() resp = await s.list_tools() self._tools = [t.name for t in resp.tools] return self._tools async def call_tool(self, name, args, trace_id, timeout_s=25, retries=1): last = None for attempt in range(retries + 1): t0 = time.time() try: async with self._open() as (read, write, _): async with ClientSession(read, write) as s: await s.initialize() out = await asyncio.wait_for(s.call_tool(name, args), timeout=timeout_s) dt = time.time() - t0 meta = {"trace_id": trace_id, "tool": name, "args": args, "latency_s": round(dt, 3)} return out, meta except Exception as e: last = e if attempt < retries: await asyncio.sleep(0.25) raise last How this works: streamablehttp_client(self._url()) opens a streamable HTTP connection to the MCP server URL. The URL includes your API key as a query param, so the server can authenticate. list_tools() is just a convenience. It asks the server which tools exist and caches the names in memory so you don’t fetch them repeatedly. call_tool() is the workhorse. It opens a session, initializes it, calls a tool with call_tool(name, args), and wraps the result with a meta object. That meta object is important later. It lets you trace which tool was called, with which params, how long it took, and which request it belonged to (trace_id). streamablehttp_client(self._url()) opens a streamable HTTP connection to the MCP server URL. The URL includes your API key as a query param, so the server can authenticate. streamablehttp_client(self._url()) list_tools() is just a convenience. It asks the server which tools exist and caches the names in memory so you don’t fetch them repeatedly. list_tools() call_tool() is the workhorse. It opens a session, initializes it, calls a tool with call_tool(name, args), and wraps the result with a meta object. call_tool() call_tool(name, args) meta That meta object is important later. It lets you trace which tool was called, with which params, how long it took, and which request it belonged to (trace_id). meta trace_id Next, we’ll build the core runner in core.py. This is where we parse the user’s request, fetch prices and fundamentals via MCP, compute metrics in Python, and then hand the facts to the LLM for narration. core.py Step 2. The assistant core (core.py) core.py This is where the assistant actually becomes “real”. client.py was just a connector. Here we decide what data to fetch, how much to fetch, how to compute the numbers, and what we hand to the model for narration. client.py i. Budgets and trace logging When you build anything that calls real tools, you want limits. Not because you don’t trust your code. Because without limits, one messy prompt can easily turn into an expensive, slow request. In our case, we cap: how far back we’ll fetch data (MAX_LOOKBACK_DAYS) how many tool calls we allow per request (MAX_TOOL_CALLS) how many tickers we’ll accept in one query (MAX_TICKERS) how far back we’ll fetch data (MAX_LOOKBACK_DAYS) MAX_LOOKBACK_DAYS how many tool calls we allow per request (MAX_TOOL_CALLS) MAX_TOOL_CALLS how many tickers we’ll accept in one query (MAX_TICKERS) MAX_TICKERS And we log a few events so we can always debug what happened later. Here’s the top part of core.py for that. core.py import json import re import time import uuid from datetime import date, timedelta from openai import OpenAI import numpy as np import pandas as pd import asyncio from client import EODHDMCP EODHD_API_KEY = "YOUR EODHD API KEY" MCP_BASE_URL = "https://mcp.eodhd.dev/mcp" MAX_LOOKBACK_DAYS = 365 MAX_TOOL_CALLS = 6 MAX_TICKERS = 5 mcp = EODHDMCP(EODHD_API_KEY, base_url=MCP_BASE_URL) oa = OpenAI(api_key = "OPENAI API KEY") NARRATION_MODEL = "gpt-5.3-chat-latest" def log_event(event, trace_id, **k): payload = {"event": event, "trace_id": trace_id, "ts": round(time.time(), 3)} payload.update(k) print(json.dumps(payload, default=str)) import json import re import time import uuid from datetime import date, timedelta from openai import OpenAI import numpy as np import pandas as pd import asyncio from client import EODHDMCP EODHD_API_KEY = "YOUR EODHD API KEY" MCP_BASE_URL = "https://mcp.eodhd.dev/mcp" MAX_LOOKBACK_DAYS = 365 MAX_TOOL_CALLS = 6 MAX_TICKERS = 5 mcp = EODHDMCP(EODHD_API_KEY, base_url=MCP_BASE_URL) oa = OpenAI(api_key = "OPENAI API KEY") NARRATION_MODEL = "gpt-5.3-chat-latest" def log_event(event, trace_id, **k): payload = {"event": event, "trace_id": trace_id, "ts": round(time.time(), 3)} payload.update(k) print(json.dumps(payload, default=str)) What’s going on here: MAX_LOOKBACK_DAYS, MAX_TOOL_CALLS, MAX_TICKERS are basically your safety rails. We’ll enforce them later, right after parsing the user query. trace_id is a small id we generate per request. Every log line includes it, so when something breaks, you can reconstruct the exact flow for that request. log_event() prints one JSON line. Nothing fancy. But it’s enough for debugging and it also looks very similar to how real systems emit traces. MAX_LOOKBACK_DAYS, MAX_TOOL_CALLS, MAX_TICKERS are basically your safety rails. We’ll enforce them later, right after parsing the user query. MAX_LOOKBACK_DAYS MAX_TOOL_CALLS MAX_TICKERS trace_id is a small id we generate per request. Every log line includes it, so when something breaks, you can reconstruct the exact flow for that request. trace_id log_event() prints one JSON line. Nothing fancy. But it’s enough for debugging and it also looks very similar to how real systems emit traces. log_event() Note: Make sure to replace YOUR EODHD API KEY with your actual EODHD API key. If you don’t have one, you can obtain it by creating an EODHD developer account. YOUR EODHD API KEY ii. Parsing the request This part is intentionally not “smart”. We’re not doing NLP. We’re not letting the model interpret the query. We just want to extract two things in a predictable way: tickers lookback window tickers lookback window That’s it. The benefit of keeping it dumb is that the behavior is stable. If the query is messy, we still do something consistent, and the rest of the pipeline remains controllable. Here are the two functions. def parse_request(text): t = (text or "").upper() raw = re.findall(r"\b[A-Z]{1,5}\b", t) bad = { "I","A","AN","THE","AND","OR","TO","FOR","OF","IN","ON","BY","WITH","ME","WE","US", "GIVE","DAY","DAYS","BRIEF","COMPARE","RANK","OVER","LAST","TREND","VOL","VOLATILITY", "DRAWDOWN","FLAG","RISKS","RISK","PLUS","MAX","MIN","LOOKBACK" } tickers = [] for x in raw: if x in bad: continue if len(x) MAX_LOOKBACK_DAYS: lookback_days = MAX_LOOKBACK_DAYS tickers = tickers[:MAX_TICKERS] return tickers, lookback_days def parse_request(text): t = (text or "").upper() raw = re.findall(r"\b[A-Z]{1,5}\b", t) bad = { "I","A","AN","THE","AND","OR","TO","FOR","OF","IN","ON","BY","WITH","ME","WE","US", "GIVE","DAY","DAYS","BRIEF","COMPARE","RANK","OVER","LAST","TREND","VOL","VOLATILITY", "DRAWDOWN","FLAG","RISKS","RISK","PLUS","MAX","MIN","LOOKBACK" } tickers = [] for x in raw: if x in bad: continue if len(x) MAX_LOOKBACK_DAYS: lookback_days = MAX_LOOKBACK_DAYS tickers = tickers[:MAX_TICKERS] return tickers, lookback_days How to read this: re.findall(r"\b[A-Z]{1,5}\b", t) pulls out every short uppercase token. That’s our crude “ticker candidate” list. The bad set is just a blacklist of common words that show up in prompts but are obviously not tickers. We keep unique tickers in order, because the first ticker becomes the “base” for correlation in the watchlist demo. Lookback is simple. Default is 30 days. If the query contains “last …”, we grab the first number after “LAST”. That avoids regex edge cases with punctuation. re.findall(r"\b[A-Z]{1,5}\b", t) pulls out every short uppercase token. That’s our crude “ticker candidate” list. re.findall(r"\b[A-Z]{1,5}\b", t) The bad set is just a blacklist of common words that show up in prompts but are obviously not tickers. bad We keep unique tickers in order, because the first ticker becomes the “base” for correlation in the watchlist demo. Lookback is simple. Default is 30 days. If the query contains “last …”, we grab the first number after “LAST”. That avoids regex edge cases with punctuation. Then enforce_budgets() clamps everything so one request can’t ask for 500 tickers or a 10-year window. enforce_budgets() Next, we’ll wire these parsed values into a request state and start making actual MCP calls for prices and fundamentals. iii. Tool wrappers: Prices and fundamentals Now we’re at the point where the assistant actually touches data. These two functions do the same job in different ways: fetch_prices() calls the MCP tool for historical prices, then normalizes the output into a tiny DataFrame with just date and close. fetch_fundamentals() calls the MCP tool for fundamentals and returns the JSON. fetch_prices() calls the MCP tool for historical prices, then normalizes the output into a tiny DataFrame with just date and close. fetch_prices() historical prices date close fetch_fundamentals() calls the MCP tool for fundamentals and returns the JSON. fetch_fundamentals() fundamentals We also keep a small state object per request. It tracks tool calls and keeps a trace of what was called. That’s how we later produce the data_used block in the final response. state data_used Here’s the code: def new_state(): return {"tool_calls": 0, "tool_trace": [], "rows": {}} def _bump(state, meta): state["tool_calls"] += 1 state["tool_trace"].append(meta) if state["tool_calls"] > MAX_TOOL_CALLS: raise RuntimeError("tool call budget exceeded") def _as_json_text(out): if isinstance(out, str): return out if hasattr(out, "content"): try: return out.content[0].text except Exception: pass return str(out) async def fetch_prices(ticker, start_date, end_date, trace_id, state): args = { "ticker": ticker, "start_date": start_date, "end_date": end_date, "period": "d", "order": "a", "fmt": "json", } out, meta = await mcp.call_tool("get_historical_stock_prices", args, trace_id) txt = _as_json_text(out) _bump(state, meta) data = json.loads(txt) if isinstance(data, dict) and data.get("error"): raise RuntimeError(data["error"]) df = pd.DataFrame(data) if df.empty: return df cols = [c for c in ["date", "close"] if c in df.columns] df = df[cols].copy() df["ticker"] = ticker state["rows"][f"{meta['tool']}:{ticker}"] = len(df) return df async def fetch_fundamentals(ticker, trace_id, state): args = { "ticker": ticker, "include_financials": False, "fmt": "json", } out, meta = await mcp.call_tool("get_fundamentals_data", args, trace_id) txt = _as_json_text(out) _bump(state, meta) data = json.loads(txt) if isinstance(data, dict) and data.get("error"): raise RuntimeError(data["error"]) return data def new_state(): return {"tool_calls": 0, "tool_trace": [], "rows": {}} def _bump(state, meta): state["tool_calls"] += 1 state["tool_trace"].append(meta) if state["tool_calls"] > MAX_TOOL_CALLS: raise RuntimeError("tool call budget exceeded") def _as_json_text(out): if isinstance(out, str): return out if hasattr(out, "content"): try: return out.content[0].text except Exception: pass return str(out) async def fetch_prices(ticker, start_date, end_date, trace_id, state): args = { "ticker": ticker, "start_date": start_date, "end_date": end_date, "period": "d", "order": "a", "fmt": "json", } out, meta = await mcp.call_tool("get_historical_stock_prices", args, trace_id) txt = _as_json_text(out) _bump(state, meta) data = json.loads(txt) if isinstance(data, dict) and data.get("error"): raise RuntimeError(data["error"]) df = pd.DataFrame(data) if df.empty: return df cols = [c for c in ["date", "close"] if c in df.columns] df = df[cols].copy() df["ticker"] = ticker state["rows"][f"{meta['tool']}:{ticker}"] = len(df) return df async def fetch_fundamentals(ticker, trace_id, state): args = { "ticker": ticker, "include_financials": False, "fmt": "json", } out, meta = await mcp.call_tool("get_fundamentals_data", args, trace_id) txt = _as_json_text(out) _bump(state, meta) data = json.loads(txt) if isinstance(data, dict) and data.get("error"): raise RuntimeError(data["error"]) return data What’s happening here: _bump() is the budget guard. Every time we make a tool call, we increment the counter and store the tool metadata. If we cross the budget, we fail fast. meta comes from client.py. It contains tool, args, and latency. That’s enough to trace “what did we call and how long did it take”. _as_json_text() is there because MCP tool results are not always plain strings. Sometimes it’s an object with .content. This helper just tries to extract the text cleanly. In fetch_prices(), we intentionally keep only date and close. That’s not because OHLC is useless. It’s because this tutorial’s metrics only need closes. Fewer columns means simpler code, smaller payloads, and fewer chances to break. _bump() is the budget guard. Every time we make a tool call, we increment the counter and store the tool metadata. If we cross the budget, we fail fast. _bump() meta comes from client.py. It contains tool, args, and latency. That’s enough to trace “what did we call and how long did it take”. meta client.py tool args _as_json_text() is there because MCP tool results are not always plain strings. Sometimes it’s an object with .content. This helper just tries to extract the text cleanly. _as_json_text() .content In fetch_prices(), we intentionally keep only date and close. That’s not because OHLC is useless. It’s because this tutorial’s metrics only need closes. Fewer columns means simpler code, smaller payloads, and fewer chances to break. fetch_prices() date close Next, we’ll compute the actual metrics. This is where the assistant stops being “an API caller” and starts producing something useful. iv. Deterministic metrics This is the most important design choice in the whole build. The model never computes numbers. Python does. So for every ticker, we compute a small set of metrics that are easy to explain and are actually useful in a “market brief” style output: total return over the window realized volatility (daily and annualized) max drawdown (worst peak-to-trough fall) a simple trend slope (so we can say “mild uptrend” or “downtrend” without vibes) a lightweight regime label (low, mid, high volatility) total return over the window realized volatility (daily and annualized) max drawdown (worst peak-to-trough fall) a simple trend slope (so we can say “mild uptrend” or “downtrend” without vibes) a lightweight regime label (low, mid, high volatility) Here’s the code. def compute_metrics(prices_df): if prices_df is None or prices_df.empty: return {} df = prices_df.copy() df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]).sort_values("date") close = pd.to_numeric(df["close"], errors="coerce").dropna() if close.empty: return {} rets = close.pct_change().dropna() out = {} # realized vol (daily), annualize with sqrt(252) if not rets.empty: out["vol_daily"] = float(rets.std()) out["vol_annualized"] = float(rets.std() * np.sqrt(252)) out["ret_total"] = float((close.iloc[-1] / close.iloc[0]) - 1.0) # max drawdown peak = close.cummax() dd = (close / peak) - 1.0 out["max_drawdown"] = float(dd.min()) # simple trend score logp = np.log(close.values) x = np.arange(len(logp)) if len(logp) >= 3: slope = np.polyfit(x, logp, 1)[0] out["trend_slope"] = float(slope) else: out["trend_slope"] = 0.0 # basic helpers out["n_points"] = int(len(close)) out["start_close"] = float(close.iloc[0]) out["end_close"] = float(close.iloc[-1]) return out def compute_regime(prices_df, window=20): # cheap regime label, based on rolling vol percentile if prices_df is None or prices_df.empty: return {"regime": "unknown"} df = prices_df.copy() df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]).sort_values("date") close = pd.to_numeric(df["close"], errors="coerce").dropna() if close.empty: return {"regime": "unknown"} rets = close.pct_change() rv = rets.rolling(window).std() last = rv.dropna() if last.empty: return {"regime": "unknown"} cur = float(last.iloc[-1]) p80 = float(last.quantile(0.8)) p50 = float(last.quantile(0.5)) if cur >= p80: reg = "high_vol" elif cur >= p50: reg = "mid_vol" else: reg = "low_vol" return {"regime": reg, "rolling_vol": cur, "window": int(window)} def compute_metrics(prices_df): if prices_df is None or prices_df.empty: return {} df = prices_df.copy() df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]).sort_values("date") close = pd.to_numeric(df["close"], errors="coerce").dropna() if close.empty: return {} rets = close.pct_change().dropna() out = {} # realized vol (daily), annualize with sqrt(252) if not rets.empty: out["vol_daily"] = float(rets.std()) out["vol_annualized"] = float(rets.std() * np.sqrt(252)) out["ret_total"] = float((close.iloc[-1] / close.iloc[0]) - 1.0) # max drawdown peak = close.cummax() dd = (close / peak) - 1.0 out["max_drawdown"] = float(dd.min()) # simple trend score logp = np.log(close.values) x = np.arange(len(logp)) if len(logp) >= 3: slope = np.polyfit(x, logp, 1)[0] out["trend_slope"] = float(slope) else: out["trend_slope"] = 0.0 # basic helpers out["n_points"] = int(len(close)) out["start_close"] = float(close.iloc[0]) out["end_close"] = float(close.iloc[-1]) return out def compute_regime(prices_df, window=20): # cheap regime label, based on rolling vol percentile if prices_df is None or prices_df.empty: return {"regime": "unknown"} df = prices_df.copy() df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]).sort_values("date") close = pd.to_numeric(df["close"], errors="coerce").dropna() if close.empty: return {"regime": "unknown"} rets = close.pct_change() rv = rets.rolling(window).std() last = rv.dropna() if last.empty: return {"regime": "unknown"} cur = float(last.iloc[-1]) p80 = float(last.quantile(0.8)) p50 = float(last.quantile(0.5)) if cur >= p80: reg = "high_vol" elif cur >= p50: reg = "mid_vol" else: reg = "low_vol" return {"regime": reg, "rolling_vol": cur, "window": int(window)} How to think about these calculations: Total return is just end / start - 1. It’s the simplest “did it go up or down” number. Volatility here is realized volatility of daily returns. That’s just the standard deviation of daily % changes. We annualize it using sqrt(252) because markets have roughly 252 trading days. Max drawdown tells you how bad the worst dip was during the window. It’s often more meaningful than return when you’re writing a quick risk note. Trend slope is intentionally simple. We fit a straight line to log prices. If the slope is positive, it’s generally drifting up. If it’s negative, it’s drifting down. Regime label is not a fancy model. It just says “compared to its own recent rolling volatility, are we currently in a high, medium, or low vol phase”. Total return is just end / start - 1. It’s the simplest “did it go up or down” number. Total return end / start - 1 Volatility here is realized volatility of daily returns. That’s just the standard deviation of daily % changes. We annualize it using sqrt(252) because markets have roughly 252 trading days. Volatility sqrt(252) Max drawdown tells you how bad the worst dip was during the window. It’s often more meaningful than return when you’re writing a quick risk note. Max drawdown Trend slope is intentionally simple. We fit a straight line to log prices. If the slope is positive, it’s generally drifting up. If it’s negative, it’s drifting down. Trend slope Regime label is not a fancy model. It just says “compared to its own recent rolling volatility, are we currently in a high, medium, or low vol phase”. Regime label The main point is this. These numbers are deterministic. If the assistant says “max drawdown was -13%”, you can trace it back to the exact close series that produced it. Next, we’ll handle the watchlist side. That means aligning returns across tickers, computing correlation, and generating a ranked snapshot. v. Watchlist utilities Once you have more than one ticker, you want two extra things: a quick ranking so you can say “this is the riskiest name in the basket” a correlation snapshot so you can see what’s moving together a quick ranking so you can say “this is the riskiest name in the basket” a correlation snapshot so you can see what’s moving together The only “gotcha” with correlation is dates. If TSLA has 41 price points and NVDA has 39 because of missing days, you can’t just correlate blindly. You need the returns lined up on the same dates first. That’s what align_returns() does. align_returns() Here’s the code: def align_returns(price_frames): if not price_frames: return pd.DataFrame() parts = [] for df in price_frames: if df is None or df.empty: continue x = df.copy() x["date"] = pd.to_datetime(x["date"], errors="coerce") x = x.dropna(subset=["date"]) x["close"] = pd.to_numeric(x["close"], errors="coerce") x = x.dropna(subset=["close"]) x = x.sort_values("date") x["ret"] = x["close"].pct_change() x = x.dropna(subset=["ret"]) parts.append(x[["date", "ticker", "ret"]]) if not parts: return pd.DataFrame() allr = pd.concat(parts, ignore_index=True) wide = allr.pivot(index="date", columns="ticker", values="ret").dropna(how="any") return wide def corr_summary(ret_wide, base_ticker, top_n=3): if ret_wide is None or ret_wide.empty: return [] if base_ticker not in ret_wide.columns: return [] c = ret_wide.corr()[base_ticker].dropna() c = c.drop(labels=[base_ticker], errors="ignore") if c.empty: return [] out = [] for k, v in c.sort_values(ascending=False).head(top_n).items(): out.append({"ticker": k, "corr": float(v)}) return out def rank_watchlist(metrics_by_ticker): rows = [] for t, m in metrics_by_ticker.items(): if not m: continue rows.append({ "ticker": t, "vol_annualized": m.get("vol_annualized"), "max_drawdown": m.get("max_drawdown"), "ret_total": m.get("ret_total"), "trend_slope": m.get("trend_slope"), }) if not rows: return pd.DataFrame() df = pd.DataFrame(rows) df = df.sort_values(["vol_annualized", "max_drawdown"], ascending=[False, True]) return df.reset_index(drop=True) def align_returns(price_frames): if not price_frames: return pd.DataFrame() parts = [] for df in price_frames: if df is None or df.empty: continue x = df.copy() x["date"] = pd.to_datetime(x["date"], errors="coerce") x = x.dropna(subset=["date"]) x["close"] = pd.to_numeric(x["close"], errors="coerce") x = x.dropna(subset=["close"]) x = x.sort_values("date") x["ret"] = x["close"].pct_change() x = x.dropna(subset=["ret"]) parts.append(x[["date", "ticker", "ret"]]) if not parts: return pd.DataFrame() allr = pd.concat(parts, ignore_index=True) wide = allr.pivot(index="date", columns="ticker", values="ret").dropna(how="any") return wide def corr_summary(ret_wide, base_ticker, top_n=3): if ret_wide is None or ret_wide.empty: return [] if base_ticker not in ret_wide.columns: return [] c = ret_wide.corr()[base_ticker].dropna() c = c.drop(labels=[base_ticker], errors="ignore") if c.empty: return [] out = [] for k, v in c.sort_values(ascending=False).head(top_n).items(): out.append({"ticker": k, "corr": float(v)}) return out def rank_watchlist(metrics_by_ticker): rows = [] for t, m in metrics_by_ticker.items(): if not m: continue rows.append({ "ticker": t, "vol_annualized": m.get("vol_annualized"), "max_drawdown": m.get("max_drawdown"), "ret_total": m.get("ret_total"), "trend_slope": m.get("trend_slope"), }) if not rows: return pd.DataFrame() df = pd.DataFrame(rows) df = df.sort_values(["vol_annualized", "max_drawdown"], ascending=[False, True]) return df.reset_index(drop=True) What’s happening here: align_returns() takes a list of price DataFrames, computes daily returns for each, then pivots them into a wide table like: date -> TSLA.US, NVDA.US, AMZN.US We drop rows where any ticker is missing, because correlation only makes sense when the returns are aligned on the same dates. corr_summary() is a compact “who moves with whom” helper. We pick one base ticker, compute correlations against everything else, then grab the top few. For a watchlist widget, that’s usually enough. rank_watchlist() is the ranking logic for the snapshot. We sort primarily by annualized volatility, and use drawdown as a secondary risk indicator. You could choose different ranking logic. The point is to keep it deterministic and explainable. align_returns() takes a list of price DataFrames, computes daily returns for each, then pivots them into a wide table like: date -> TSLA.US, NVDA.US, AMZN.US align_returns() date -> TSLA.US, NVDA.US, AMZN.US We drop rows where any ticker is missing, because correlation only makes sense when the returns are aligned on the same dates. corr_summary() is a compact “who moves with whom” helper. We pick one base ticker, compute correlations against everything else, then grab the top few. For a watchlist widget, that’s usually enough. corr_summary() rank_watchlist() is the ranking logic for the snapshot. We sort primarily by annualized volatility, and use drawdown as a secondary risk indicator. You could choose different ranking logic. The point is to keep it deterministic and explainable. rank_watchlist() Next, we’ll build the facts objects and narration layer. That’s where we enforce the “model is just a narrator” contract. vi. Facts object and narration This is where the “narrator pattern” becomes real. Up to this point, we’ve done everything with MCP and Python. We fetched prices and fundamentals from EODHD, we computed metrics, we aligned returns. Now we need one clean object that represents “the truth” for this request. That’s what the facts object is. facts The rule is simple. facts contains only things we actually fetched or computed. The model never sees raw market data. It sees the cleaned facts. The model is told to write using only those facts, and not to invent any numbers. facts contains only things we actually fetched or computed. facts The model never sees raw market data. It sees the cleaned facts. The model is told to write using only those facts, and not to invent any numbers. Here are the functions that build those facts objects for the two demos, plus the narration function. def build_facts_single(ticker, lookback_days, metrics, regime, fundamentals): # keep this compact. LLM will narrate from this later out = { "type": "single_ticker_brief", "ticker": ticker, "lookback_days": int(lookback_days), "metrics": metrics, "regime": regime, } if isinstance(fundamentals, dict): gen = fundamentals.get("General", {}) or {} highlights = { "name": gen.get("Name"), "exchange": gen.get("Exchange"), "sector": gen.get("Sector"), "industry": gen.get("Industry"), "market_cap": gen.get("MarketCapitalization"), "pe": gen.get("PERatio"), "beta": gen.get("Beta"), "div_yield": gen.get("DividendYield"), } out["fundamentals"] = {k: v for k, v in highlights.items() if v is not None} return out def build_facts_watchlist(tickers, lookback_days, rank_df, corr_bits, metrics_by_ticker): out = { "type": "watchlist_snapshot", "tickers": tickers, "lookback_days": int(lookback_days), "ranking": rank_df.to_dict(orient="records") if isinstance(rank_df, pd.DataFrame) else [], "correlation": corr_bits, "metrics_by_ticker": metrics_by_ticker, } return out def narrate(facts): prompt = ( "Write a short, product-ready market note using ONLY the facts below.\n" "No guessing. No extra numbers. If something is missing, say it's missing.\n" "Keep it tight and readable.\n\n" f"FACTS:\n{json.dumps(facts, indent=2, default=str)}" ) r = oa.responses.create( model=NARRATION_MODEL, input=[{"role": "user", "content": prompt}], ) try: return r.output_text except Exception: return str(r) def build_facts_single(ticker, lookback_days, metrics, regime, fundamentals): # keep this compact. LLM will narrate from this later out = { "type": "single_ticker_brief", "ticker": ticker, "lookback_days": int(lookback_days), "metrics": metrics, "regime": regime, } if isinstance(fundamentals, dict): gen = fundamentals.get("General", {}) or {} highlights = { "name": gen.get("Name"), "exchange": gen.get("Exchange"), "sector": gen.get("Sector"), "industry": gen.get("Industry"), "market_cap": gen.get("MarketCapitalization"), "pe": gen.get("PERatio"), "beta": gen.get("Beta"), "div_yield": gen.get("DividendYield"), } out["fundamentals"] = {k: v for k, v in highlights.items() if v is not None} return out def build_facts_watchlist(tickers, lookback_days, rank_df, corr_bits, metrics_by_ticker): out = { "type": "watchlist_snapshot", "tickers": tickers, "lookback_days": int(lookback_days), "ranking": rank_df.to_dict(orient="records") if isinstance(rank_df, pd.DataFrame) else [], "correlation": corr_bits, "metrics_by_ticker": metrics_by_ticker, } return out def narrate(facts): prompt = ( "Write a short, product-ready market note using ONLY the facts below.\n" "No guessing. No extra numbers. If something is missing, say it's missing.\n" "Keep it tight and readable.\n\n" f"FACTS:\n{json.dumps(facts, indent=2, default=str)}" ) r = oa.responses.create( model=NARRATION_MODEL, input=[{"role": "user", "content": prompt}], ) try: return r.output_text except Exception: return str(r) What’s happening here: build_facts_single() takes the ticker, window, computed metrics, the vol regime label, and the fundamentals payload. But it doesn’t dump the entire fundamentals JSON. It picks a handful of fields from the General section and only keeps what exists. That keeps the prompt tight and the output predictable. build_facts_watchlist() is the same idea but for multiple tickers. It passes the ranking table, correlation notes, and per-ticker metrics. narrate() is basically “convert this facts object into human-friendly text”. The prompt is strict on purpose. If the model can only see these facts, it cannot hallucinate numbers outside them. build_facts_single() takes the ticker, window, computed metrics, the vol regime label, and the fundamentals payload. But it doesn’t dump the entire fundamentals JSON. It picks a handful of fields from the General section and only keeps what exists. That keeps the prompt tight and the output predictable. build_facts_single() General build_facts_watchlist() is the same idea but for multiple tickers. It passes the ranking table, correlation notes, and per-ticker metrics. build_facts_watchlist() narrate() is basically “convert this facts object into human-friendly text”. The prompt is strict on purpose. If the model can only see these facts, it cannot hallucinate numbers outside them. narrate() One small implementation detail. narrate() is a normal blocking function, while everything else is async. That’s why later, inside run_assistant(), we call it with await asyncio.to_thread(...) so it doesn’t block the async flow. narrate() run_assistant() await asyncio.to_thread(...) vii. The orchestration function (run_assistant()) run_assistant() This is the piece that ties everything together. It does four things in order: create a trace id and log the request parse tickers and lookback, then clamp them to budgets fetch EODHD data via MCP and compute metrics in Python call the model to narrate the facts, then return a structured response create a trace id and log the request parse tickers and lookback, then clamp them to budgets fetch EODHD data via MCP and compute metrics in Python call the model to narrate the facts, then return a structured response Here’s the function: def _dates_from_lookback(lookback_days): end = date.today() start = end - timedelta(days=int(lookback_days)) return start.isoformat(), end.isoformat() async def run_assistant(user_text, mode="auto"): trace_id = uuid.uuid4().hex[:10] log_event("request_started", trace_id, text=user_text, mode=mode) tickers, lookback = parse_request(user_text) tickers, lookback = enforce_budgets(tickers, lookback) if not tickers: return { "answer": "no tickers found in request", "metrics": {}, "data_used": {}, "tool_trace_id": trace_id, } log_event("parsed", trace_id, tickers=tickers, lookback_days=lookback) start_date, end_date = _dates_from_lookback(lookback) state = new_state() if mode == "auto": mode = "watchlist" if len(tickers) > 1 else "single" try: if mode == "single": t = tickers[0] t_full = t if "." in t else f"{t}.US" log_event("tool_phase", trace_id, mode="single", ticker=t_full, start_date=start_date, end_date=end_date) prices = await fetch_prices(t_full, start_date, end_date, trace_id, state) metrics = compute_metrics(prices) regime = compute_regime(prices) fundamentals = await fetch_fundamentals(t_full, trace_id, state) facts = build_facts_single(t_full, lookback, metrics, regime, fundamentals) answer = await asyncio.to_thread(narrate, facts) resp = { "answer": answer, "metrics": metrics, "data_used": { "tickers": [t_full], "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } log_event("request_finished", trace_id, tool_calls=state["tool_calls"]) return resp # watchlist full = [x if "." in x else f"{x}.US" for x in tickers] log_event("tool_phase", trace_id, mode="watchlist", tickers=full, start_date=start_date, end_date=end_date) frames = [] metrics_by = {} for t in full: prices = await fetch_prices(t, start_date, end_date, trace_id, state) frames.append(prices) metrics_by[t] = compute_metrics(prices) ret_wide = align_returns(frames) base = full[0] corr_bits = [] top = corr_summary(ret_wide, base, top_n=3) if top: corr_bits.append({"base": base, "top": top}) rank_df = rank_watchlist(metrics_by) facts = build_facts_watchlist(full, lookback, rank_df, corr_bits, metrics_by) answer = await asyncio.to_thread(narrate, facts) resp = { "answer": answer, "metrics": {"by_ticker": metrics_by}, "data_used": { "tickers": full, "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } log_event("request_finished", trace_id, tool_calls=state["tool_calls"]) return resp except Exception as e: detail = repr(e) if hasattr(e, "exceptions"): detail = detail + " | " + " ; ".join([repr(x) for x in e.exceptions]) log_event("request_failed", trace_id, err=detail) return { "answer": f"failed: {e}", "metrics": {}, "data_used": { "tickers": tickers, "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } def _dates_from_lookback(lookback_days): end = date.today() start = end - timedelta(days=int(lookback_days)) return start.isoformat(), end.isoformat() async def run_assistant(user_text, mode="auto"): trace_id = uuid.uuid4().hex[:10] log_event("request_started", trace_id, text=user_text, mode=mode) tickers, lookback = parse_request(user_text) tickers, lookback = enforce_budgets(tickers, lookback) if not tickers: return { "answer": "no tickers found in request", "metrics": {}, "data_used": {}, "tool_trace_id": trace_id, } log_event("parsed", trace_id, tickers=tickers, lookback_days=lookback) start_date, end_date = _dates_from_lookback(lookback) state = new_state() if mode == "auto": mode = "watchlist" if len(tickers) > 1 else "single" try: if mode == "single": t = tickers[0] t_full = t if "." in t else f"{t}.US" log_event("tool_phase", trace_id, mode="single", ticker=t_full, start_date=start_date, end_date=end_date) prices = await fetch_prices(t_full, start_date, end_date, trace_id, state) metrics = compute_metrics(prices) regime = compute_regime(prices) fundamentals = await fetch_fundamentals(t_full, trace_id, state) facts = build_facts_single(t_full, lookback, metrics, regime, fundamentals) answer = await asyncio.to_thread(narrate, facts) resp = { "answer": answer, "metrics": metrics, "data_used": { "tickers": [t_full], "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } log_event("request_finished", trace_id, tool_calls=state["tool_calls"]) return resp # watchlist full = [x if "." in x else f"{x}.US" for x in tickers] log_event("tool_phase", trace_id, mode="watchlist", tickers=full, start_date=start_date, end_date=end_date) frames = [] metrics_by = {} for t in full: prices = await fetch_prices(t, start_date, end_date, trace_id, state) frames.append(prices) metrics_by[t] = compute_metrics(prices) ret_wide = align_returns(frames) base = full[0] corr_bits = [] top = corr_summary(ret_wide, base, top_n=3) if top: corr_bits.append({"base": base, "top": top}) rank_df = rank_watchlist(metrics_by) facts = build_facts_watchlist(full, lookback, rank_df, corr_bits, metrics_by) answer = await asyncio.to_thread(narrate, facts) resp = { "answer": answer, "metrics": {"by_ticker": metrics_by}, "data_used": { "tickers": full, "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } log_event("request_finished", trace_id, tool_calls=state["tool_calls"]) return resp except Exception as e: detail = repr(e) if hasattr(e, "exceptions"): detail = detail + " | " + " ; ".join([repr(x) for x in e.exceptions]) log_event("request_failed", trace_id, err=detail) return { "answer": f"failed: {e}", "metrics": {}, "data_used": { "tickers": tickers, "date_range": [start_date, end_date], "tools_called": [x.get("tool") for x in state["tool_trace"]], "tool_calls": state["tool_calls"], }, "tool_trace_id": trace_id, } This function is the glue. It creates a trace_id, logs the request, extracts tickers and a lookback window, then clamps both to your budgets so the assistant can’t over-fetch or spam tool calls. trace_id After that, it turns the lookback into a start_date and end_date, initializes a fresh state, and picks a mode. In single mode it fetches prices and fundamentals for one ticker via EODHD’s MCP tools, computes the metrics in Python, packs everything into a facts object, and asks the LLM to only narrate those facts. In watchlist mode it does the same across multiple tickers, then aligns returns so correlation is computed on matching dates, and builds a ranked snapshot. start_date end_date state single watchlist The response is always structured the same way. You get the narrative answer, the raw computed metrics, a data_used block that shows tickers, date range, and tools called, plus a tool_trace_id so you can trace any output back to logs. answer metrics data_used tool_trace_id That structure is the difference between “a chat response” and “a shippable assistant output”. You can plug the same response into a UI card, a Slack alert, or a dashboard without changing anything. Demo 1. Market brief for one ticker Let’s start with the simplest flow. One ticker, one lookback window, and a market brief that looks like something you could show inside a product. Prompt used: Prompt used: “Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights.” “Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights.” Code (Jupyter Notebook): Code (Jupyter Notebook): import asyncio import json from core import run_assistant q1 = "Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights." r1 = await run_assistant(q1, mode="single") print(json.dumps(r1, indent=2, ensure_ascii=False)) import asyncio import json from core import run_assistant q1 = "Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights." r1 = await run_assistant(q1, mode="single") print(json.dumps(r1, indent=2, ensure_ascii=False)) Output: Output: {"event": "request_started", "trace_id": "2af550173f", "ts": 1772735388.777, "text": "Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights.", "mode": "single"} {"event": "parsed", "trace_id": "2af550173f", "ts": 1772735388.778, "tickers": ["AAPL"], "lookback_days": 30} {"event": "tool_phase", "trace_id": "2af550173f", "ts": 1772735388.778, "mode": "single", "ticker": "AAPL.US", "start_date": "2026-02-03", "end_date": "2026-03-05"} {"event": "request_finished", "trace_id": "2af550173f", "ts": 1772735404.392, "tool_calls": 2} { "answer": "Apple Inc (AAPL.US) | NASDAQ | Technology — Consumer Electronics\n \nOver the past 30 days, Apple shares declined 2.58%, falling from 269.48 to 262.52 across 21 trading observations. The trend slope over the period was negative (-0.00175), indicating a modest downward drift.\n\nRealized daily volatility was 1.93%, equivalent to about 30.65% annualized. The stock is currently classified in a high‑volatility regime based on a 20‑day rolling volatility measure. \n\nMaximum drawdown during the period reached -8.03%.\n\nAdditional fundamentals or valuation metrics were not provided.", "metrics": { "vol_daily": 0.01930981768788001, "vol_annualized": 0.3065338527847606, "ret_total": -0.02582751966750796, "max_drawdown": -0.08032503955127279, "trend_slope": -0.0017498633497641184, "n_points": 21, "start_close": 269.48, "end_close": 262.52 }, "data_used": { "tickers": [ "AAPL.US" ], "date_range": [ "2026-02-03", "2026-03-05" ], "tools_called": [ "get_historical_stock_prices", "get_fundamentals_data" ], "tool_calls": 2 }, "tool_trace_id": "2af550173f" } {"event": "request_started", "trace_id": "2af550173f", "ts": 1772735388.777, "text": "Give me a 30-day brief for AAPL. trend, volatility, max drawdown, plus 3 fundamental highlights.", "mode": "single"} {"event": "parsed", "trace_id": "2af550173f", "ts": 1772735388.778, "tickers": ["AAPL"], "lookback_days": 30} {"event": "tool_phase", "trace_id": "2af550173f", "ts": 1772735388.778, "mode": "single", "ticker": "AAPL.US", "start_date": "2026-02-03", "end_date": "2026-03-05"} {"event": "request_finished", "trace_id": "2af550173f", "ts": 1772735404.392, "tool_calls": 2} { "answer": "Apple Inc (AAPL.US) | NASDAQ | Technology — Consumer Electronics\n \nOver the past 30 days, Apple shares declined 2.58%, falling from 269.48 to 262.52 across 21 trading observations. The trend slope over the period was negative (-0.00175), indicating a modest downward drift.\n\nRealized daily volatility was 1.93%, equivalent to about 30.65% annualized. The stock is currently classified in a high‑volatility regime based on a 20‑day rolling volatility measure. \n\nMaximum drawdown during the period reached -8.03%.\n\nAdditional fundamentals or valuation metrics were not provided.", "metrics": { "vol_daily": 0.01930981768788001, "vol_annualized": 0.3065338527847606, "ret_total": -0.02582751966750796, "max_drawdown": -0.08032503955127279, "trend_slope": -0.0017498633497641184, "n_points": 21, "start_close": 269.48, "end_close": 262.52 }, "data_used": { "tickers": [ "AAPL.US" ], "date_range": [ "2026-02-03", "2026-03-05" ], "tools_called": [ "get_historical_stock_prices", "get_fundamentals_data" ], "tool_calls": 2 }, "tool_trace_id": "2af550173f" } First, you’ll see the log events. They’re not part of the final response. They’re just the trace trail. request_started shows the raw prompt and that we forced mode="single". parsed confirms the parser extracted AAPL and a 30-day lookback. tool_phase shows what we actually fetched: AAPL.US from 2026-02-03 to 2026-03-05. request_finished confirms we made exactly 2 tool calls. request_started shows the raw prompt and that we forced mode="single". request_started mode="single" parsed confirms the parser extracted AAPL and a 30-day lookback. parsed AAPL tool_phase shows what we actually fetched: AAPL.US from 2026-02-03 to 2026-03-05. tool_phase AAPL.US 2026-02-03 2026-03-05 request_finished confirms we made exactly 2 tool calls. request_finished 2 tool calls. Now the actual response JSON: answer is the narrative. In this run it summarizes: answer return of -2.58% (269.48 to 262.52) 21 price observations in that window negative trend slope (-0.00175) meaning mild downward drift daily vol 1.93% and annualized vol 30.65% max drawdown -8.03% and it labels the regime as high volatility using the rolling vol logic. return of -2.58% (269.48 to 262.52) -2.58% 21 price observations in that window 21 negative trend slope (-0.00175) meaning mild downward drift -0.00175 daily vol 1.93% and annualized vol 30.65% 1.93% 30.65% max drawdown -8.03% -8.03% and it labels the regime as high volatility using the rolling vol logic. metrics is where those numbers come from. This is the deterministic part. ret_total, vol_daily, vol_annualized, max_drawdown, and trend_slope were computed directly from the fetched closes. start_close, end_close, and n_points explain the exact series used. metrics ret_total vol_daily vol_annualized max_drawdown trend_slope start_close end_close n_points data_used is the audit block for this specific output. It shows: data_used ticker normalized to AAPL.US the exact date range pulled the exact MCP tools used: get_historical_stock_prices and get_fundamentals_data and again, tool_calls: 2 so you can quickly spot runaway calls. ticker normalized to AAPL.US AAPL.US the exact date range pulled the exact MCP tools used: get_historical_stock_prices and get_fundamentals_data get_historical_stock_prices get_fundamentals_data and again, tool_calls: 2 so you can quickly spot runaway calls. tool_calls: 2 tool_trace_id (2af550173f) is your handle for debugging. Every log line above carries the same id, so you can trace this brief back to the exact tool calls and parameters. tool_trace_id 2af550173f Demo 2. Watchlist snapshot Now let’s switch to the watchlist flow. Same assistant core. The only difference is we pass multiple tickers and a longer window, so the output becomes a comparative risk snapshot. Prompt used: Prompt used: “Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks.” “Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks.” Code: Code: q2 = "Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks." r2 = await run_assistant(q2, mode="watchlist") print(json.dumps(r2, indent=2, ensure_ascii=False)) q2 = "Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks." r2 = await run_assistant(q2, mode="watchlist") print(json.dumps(r2, indent=2, ensure_ascii=False)) Output: Output: {"event": "request_started", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "text": "Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks.", "mode": "watchlist"} {"event": "parsed", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "tickers": ["TSLA", "NVDA", "AMZN"], "lookback_days": 60} {"event": "tool_phase", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "mode": "watchlist", "tickers": ["TSLA.US", "NVDA.US", "AMZN.US"], "start_date": "2026-01-05", "end_date": "2026-03-06"} {"event": "request_finished", "trace_id": "1b67bb47d6", "ts": 1772735423.004, "tool_calls": 3} { "answer": "Market Watchlist Snapshot (last 60 days)\n\nAll three names show negative total returns and downward trend slopes over the period.\n\nNVDA.US ranks highest in the group despite a small decline. Total return is -0.027. Price moved from 188.12 to 183.04 across 41 observations. Annualized volatility is 0.3808 and maximum drawdown is -0.107.\n\nTSLA.US shows the second‑highest volatility profile with annualized volatility of 0.3561. Total return is -0.101, with price falling from 451.67 to 405.94. Maximum drawdown reached -0.131. Trend slope is negative. \n\nAMZN.US has the lowest volatility in the set (annualized 0.3196) but the deepest drawdown at -0.196. Total return is -0.0697, with price moving from 233.06 to 216.82. Trend slope is also negative.\n\nCorrelation: TSLA shows a stronger relationship with NVDA (0.533) than with AMZN (0.177).\n\nMissing from the data: trading volume, catalysts, sector context, and forward-looking indicators.", "metrics": { "by_ticker": { "TSLA.US": { "vol_daily": 0.02243518393199404, "vol_annualized": 0.3561475038122908, "ret_total": -0.10124648526579139, "max_drawdown": -0.13115770363318358, "trend_slope": -0.0026452119688441023, "n_points": 41, "start_close": 451.67, "end_close": 405.94 }, "NVDA.US": { "vol_daily": 0.023987861378298222, "vol_annualized": 0.3807954941476091, "ret_total": -0.027004039974484417, "max_drawdown": -0.10716326424601319, "trend_slope": -4.3573704505466623e-05, "n_points": 41, "start_close": 188.12, "end_close": 183.04 }, "AMZN.US": { "vol_daily": 0.020129905817481322, "vol_annualized": 0.31955234824924766, "ret_total": -0.06968162704882863, "max_drawdown": -0.1964184655186353, "trend_slope": -0.00520436173926906, "n_points": 41, "start_close": 233.06, "end_close": 216.82 } } }, "data_used": { "tickers": [ "TSLA.US", "NVDA.US", "AMZN.US" ], "date_range": [ "2026-01-05", "2026-03-06" ], "tools_called": [ "get_historical_stock_prices", "get_historical_stock_prices", "get_historical_stock_prices" ], "tool_calls": 3 }, "tool_trace_id": "1b67bb47d6" } {"event": "request_started", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "text": "Compare TSLA, NVDA, AMZN over the last 60 days. rank by volatility and drawdown, and flag valuation risks.", "mode": "watchlist"} {"event": "parsed", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "tickers": ["TSLA", "NVDA", "AMZN"], "lookback_days": 60} {"event": "tool_phase", "trace_id": "1b67bb47d6", "ts": 1772735404.394, "mode": "watchlist", "tickers": ["TSLA.US", "NVDA.US", "AMZN.US"], "start_date": "2026-01-05", "end_date": "2026-03-06"} {"event": "request_finished", "trace_id": "1b67bb47d6", "ts": 1772735423.004, "tool_calls": 3} { "answer": "Market Watchlist Snapshot (last 60 days)\n\nAll three names show negative total returns and downward trend slopes over the period.\n\nNVDA.US ranks highest in the group despite a small decline. Total return is -0.027. Price moved from 188.12 to 183.04 across 41 observations. Annualized volatility is 0.3808 and maximum drawdown is -0.107.\n\nTSLA.US shows the second‑highest volatility profile with annualized volatility of 0.3561. Total return is -0.101, with price falling from 451.67 to 405.94. Maximum drawdown reached -0.131. Trend slope is negative. \n\nAMZN.US has the lowest volatility in the set (annualized 0.3196) but the deepest drawdown at -0.196. Total return is -0.0697, with price moving from 233.06 to 216.82. Trend slope is also negative.\n\nCorrelation: TSLA shows a stronger relationship with NVDA (0.533) than with AMZN (0.177).\n\nMissing from the data: trading volume, catalysts, sector context, and forward-looking indicators.", "metrics": { "by_ticker": { "TSLA.US": { "vol_daily": 0.02243518393199404, "vol_annualized": 0.3561475038122908, "ret_total": -0.10124648526579139, "max_drawdown": -0.13115770363318358, "trend_slope": -0.0026452119688441023, "n_points": 41, "start_close": 451.67, "end_close": 405.94 }, "NVDA.US": { "vol_daily": 0.023987861378298222, "vol_annualized": 0.3807954941476091, "ret_total": -0.027004039974484417, "max_drawdown": -0.10716326424601319, "trend_slope": -4.3573704505466623e-05, "n_points": 41, "start_close": 188.12, "end_close": 183.04 }, "AMZN.US": { "vol_daily": 0.020129905817481322, "vol_annualized": 0.31955234824924766, "ret_total": -0.06968162704882863, "max_drawdown": -0.1964184655186353, "trend_slope": -0.00520436173926906, "n_points": 41, "start_close": 233.06, "end_close": 216.82 } } }, "data_used": { "tickers": [ "TSLA.US", "NVDA.US", "AMZN.US" ], "date_range": [ "2026-01-05", "2026-03-06" ], "tools_called": [ "get_historical_stock_prices", "get_historical_stock_prices", "get_historical_stock_prices" ], "tool_calls": 3 }, "tool_trace_id": "1b67bb47d6" } The logs show the assistant correctly extracted TSLA, NVDA, AMZN and a 60-day lookback, then fetched TSLA.US, NVDA.US, and AMZN.US from 2026-01-05 to 2026-03-06. Since this is a watchlist request, it made exactly 3 tool calls. One get_historical_stock_prices call per ticker. TSLA NVDA AMZN 60-day TSLA.US NVDA.US AMZN.US 2026-01-05 2026-03-06 3 get_historical_stock_prices Inside answer, the model is basically summarizing what Python computed. In this run, all three names had negative returns and negative trend slopes. NVDA had the highest annualized volatility at 0.3808 with a relatively small decline of -2.7%. TSLA was next in volatility (0.3561) with a larger decline (-10.1%) and drawdown of about -13.1%. AMZN had the lowest volatility (0.3196) but the deepest drawdown at around -19.6%. It also includes a correlation note derived from the aligned returns table. TSLA’s return series correlated more with NVDA (0.533) than with AMZN (0.177) in this window. answer 0.3808 -2.7% 0.3561 -10.1% -13.1% 0.3196 -19.6% 0.533 0.177 metrics.by_ticker is where the snapshot really lives. It contains the full computed metric set per ticker, including observation count (n_points=41) and the start and end closes used for the return calculation. data_used shows exactly what we fetched, including the tickers, the date range, and the three price tool calls. And tool_trace_id is the id that links this output back to the full trace logs. metrics.by_ticker n_points=41 data_used tool_trace_id How a product team would use this. This output is already shaped like a widget backend. You can render the ranking as a watchlist “risk card”, show the top volatility and drawdown names, and drop the narrative into a compact summary box. Since you also get deterministic metrics, you can build UI elements without parsing text, and still keep the narration as a layer on top. metrics What makes this shippable, and what can be improved The core reason this works in a real product setting is that the numbers are deterministic. Prices and fundamentals come from EODHD via MCP, metrics are computed in Python, and the model only writes narrative from a facts object. On top of that, every run is traceable. You get tool logs, data_used, and a tool_trace_id, plus hard limits on lookback, tickers, and tool calls so the system can’t spiral. data_used tool_trace_id At the same time, this is still an MVP. The parsing is a simple heuristic, the metric set is intentionally small, and fundamentals are only lightly extracted. If you want to take this further, the next upgrades are straightforward. Add volume and a couple more data tools like earnings calendar and news, introduce caching for repeated requests, build a tiny evaluation harness with fixed prompts and expected outputs, then wrap run_assistant() behind a small API so it can power an actual UI or internal service. run_assistant() Conclusion The main takeaway is simple. If you want a financial assistant to be usable beyond casual chat, you need to separate facts from narrative. MCP gives you a clean way to pull real data through tools, Python gives you deterministic metrics, and the model becomes the last-mile layer that turns those facts into readable output. This is still a small build, but it’s already shaped like something you can ship. The response format is structured, traceable, and easy to plug into a UI. If you extend it with a few more tools and add basic caching, it can quickly move from a jupyter notebook demo to a real feature. If you want to try the same approach with a full market data tool layer out of the box, EODHD’s MCP server is a solid starting point. With that being said, you’ve reached the end of the article. Hope you learned something new and useful today. Thank you very much for your time.