Les agents de trading ont besoin de deux flux de données : les actualités du marché en temps réel et les données financières structurées. Ce tutoriel construit un serveur MCP qui combine la recherche Scavio pour les dernières nouvelles et le sentiment avec une API de données financières pour les prix et les fondamentaux, exposant les deux sous forme d'appels d'outils que tout agent compatible MCP peut invoquer.
Prérequis
- Python 3.11+
- Une clé API Scavio depuis https://scavio.dev
- Une clé API gratuite de données financières (Alpha Vantage ou similaire)
- Le package Python mcp (pip install mcp)
Parcours
Étape 1: Configurer le squelette du serveur MCP
Créez le serveur MCP avec le gestionnaire de protocole standard. Ce serveur exposera trois outils : market_news, stock_quote et trading_signal.
Python
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
import json
import os
SCAVIO_API_KEY = os.environ.get("SCAVIO_API_KEY", "your-api-key")
FINANCE_API_KEY = os.environ.get("FINANCE_API_KEY", "your-finance-key")
server = Server("trading-data")
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="market_news",
description="Search for breaking market news and analyst commentary on a stock or sector",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Stock ticker or sector to search"},
"num_results": {"type": "integer", "default": 5}
},
"required": ["query"]
}
),
Tool(
name="stock_quote",
description="Get current price, volume, and daily change for a stock ticker",
inputSchema={
"type": "object",
"properties": {
"ticker": {"type": "string", "description": "Stock ticker symbol"}
},
"required": ["ticker"]
}
),
Tool(
name="trading_signal",
description="Combine news sentiment with price data to generate a trading signal",
inputSchema={
"type": "object",
"properties": {
"ticker": {"type": "string"},
"timeframe": {"type": "string", "default": "1d"}
},
"required": ["ticker"]
}
)
]Étape 2: Implémenter l'outil d'actualités du marché avec la recherche Scavio
L'outil market_news recherche les actualités financières récentes, filtre le contenu pertinent pour le marché et renvoie des résumés structurés avec les URL sources.
Python
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "market_news":
return await handle_market_news(arguments)
elif name == "stock_quote":
return await handle_stock_quote(arguments)
elif name == "trading_signal":
return await handle_trading_signal(arguments)
return [TextContent(type="text", text=f"Unknown tool: {name}")]
async def handle_market_news(args: dict) -> list[TextContent]:
query = args["query"]
num_results = args.get("num_results", 5)
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={
"query": f"{query} stock market news 2026",
"num_results": num_results
}
)
resp.raise_for_status()
results = resp.json().get("results", [])
news = []
for r in results:
news.append({
"title": r.get("title", ""),
"url": r.get("url", ""),
"summary": r.get("description", "")[:300],
"source": r.get("url", "").split("/")[2] if r.get("url") else ""
})
return [TextContent(type="text", text=json.dumps({"news": news, "count": len(news)}, indent=2))]Étape 3: Implémenter les outils de cours d'action et de signal de trading
L'outil stock_quote récupère les données de prix depuis une API financière. L'outil trading_signal combine le sentiment des actualités avec les mouvements de prix pour produire un signal simple haussier/baissier/neutre.
Python
async def handle_stock_quote(args: dict) -> list[TextContent]:
ticker = args["ticker"].upper()
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
"https://www.alphavantage.co/query",
params={
"function": "GLOBAL_QUOTE",
"symbol": ticker,
"apikey": FINANCE_API_KEY
}
)
resp.raise_for_status()
data = resp.json().get("Global Quote", {})
quote = {
"ticker": ticker,
"price": data.get("05. price", "N/A"),
"change": data.get("09. change", "N/A"),
"change_pct": data.get("10. change percent", "N/A"),
"volume": data.get("06. volume", "N/A")
}
return [TextContent(type="text", text=json.dumps(quote, indent=2))]
async def handle_trading_signal(args: dict) -> list[TextContent]:
ticker = args["ticker"].upper()
# Get news
news_result = await handle_market_news({"query": ticker, "num_results": 5})
news_data = json.loads(news_result[0].text)
# Get price
quote_result = await handle_stock_quote({"ticker": ticker})
quote_data = json.loads(quote_result[0].text)
# Simple sentiment scoring from news titles
positive_words = {"surge", "jump", "rally", "beat", "upgrade", "growth", "record", "gain"}
negative_words = {"drop", "fall", "crash", "miss", "downgrade", "decline", "loss", "cut"}
sentiment_score = 0
for article in news_data.get("news", []):
title_words = set(article["title"].lower().split())
sentiment_score += len(title_words & positive_words)
sentiment_score -= len(title_words & negative_words)
# Combine with price change
change_pct = quote_data.get("change_pct", "0%").replace("%", "")
try:
price_signal = float(change_pct)
except ValueError:
price_signal = 0.0
# Generate signal
combined = sentiment_score + (1 if price_signal > 0 else -1 if price_signal < 0 else 0)
if combined >= 2:
signal = "BULLISH"
elif combined <= -2:
signal = "BEARISH"
else:
signal = "NEUTRAL"
result = {
"ticker": ticker,
"signal": signal,
"sentiment_score": sentiment_score,
"price_change": change_pct + "%",
"news_count": len(news_data.get("news", [])),
"confidence": "high" if abs(combined) >= 3 else "medium" if abs(combined) >= 2 else "low"
}
return [TextContent(type="text", text=json.dumps(result, indent=2))]Étape 4: Exécuter le serveur MCP
Démarrez le serveur MCP via stdio afin que tout client compatible MCP (Claude Desktop, Hermes, agents personnalisés) puisse se connecter et invoquer les outils de trading.
Python
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
# MCP client config (add to mcp_config.json):
# {
# "mcpServers": {
# "trading-data": {
# "command": "python",
# "args": ["trading_mcp_server.py"],
# "env": {
# "SCAVIO_API_KEY": "your-api-key",
# "FINANCE_API_KEY": "your-finance-key"
# }
# }
# }
# }Exemple Python
Python
import asyncio
import httpx
import json
SCAVIO_API_KEY = "your-api-key"
async def market_news(ticker: str) -> dict:
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": f"{ticker} stock market news 2026", "num_results": 5}
)
resp.raise_for_status()
results = resp.json().get("results", [])
return {
"ticker": ticker,
"articles": [{"title": r["title"], "url": r["url"]} for r in results],
"count": len(results)
}
async def main():
news = await market_news("NVDA")
print(f"Found {news['count']} articles for {news['ticker']}")
for a in news["articles"]:
print(f" {a['title']}")
asyncio.run(main())Exemple JavaScript
JavaScript
const SCAVIO_API_KEY = "your-api-key";
async function marketNews(ticker) {
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query: ticker + " stock market news 2026", num_results: 5 })
});
const data = await resp.json();
return {
ticker,
articles: (data.results || []).map(r => ({ title: r.title, url: r.url })),
count: (data.results || []).length
};
}
async function main() {
const news = await marketNews("NVDA");
console.log(`Found ${news.count} articles for ${news.ticker}`);
news.articles.forEach(a => console.log(" " + a.title));
}
main();Sortie attendue
JSON
Found 5 articles for NVDA
NVIDIA Q2 2026 Earnings Beat Expectations...
NVDA Stock Surges on New AI Chip Announcement...