Los agentes comerciales necesitan dos flujos de datos: noticias del mercado en tiempo real y datos financieros estructurados. Este tutorial crea un servidor MCP que combina la búsqueda de Scavio de noticias y sentimientos de última hora con una API de datos financieros para precios y fundamentos, exponiendo ambas como llamadas de herramientas que cualquier agente compatible con MCP puede invocar.
Requisitos previos
- Python 3.11+
- Una clave API de Scavio de https://scavio.dev
- Una clave API de datos financieros gratuita (Alpha Vantage o similar)
- El paquete mcp Python (pip install mcp)
Guia paso a paso
Paso 1: Configurar el esqueleto del servidor MCP
Cree el servidor MCP con el controlador de protocolo estándar. Este servidor expondrá tres herramientas: market_news, stock_quote y 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"]
}
)
]Paso 2: Implementar la herramienta de noticias del mercado con búsqueda Scavio
La herramienta market_news busca noticias financieras recientes, filtra contenido relevante para el mercado y devuelve resúmenes estructurados con URL de origen.
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))]Paso 3: Implementar las herramientas de cotización de acciones y señales comerciales
La herramienta stock_quote obtiene datos de precios de una API financiera. La herramienta trading_signal combina el sentimiento de las noticias con el movimiento de precios para producir una señal alcista/bajista/neutral simple.
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))]Paso 4: Ejecute el servidor MCP
Inicie el servidor MCP en stdio para que cualquier cliente compatible con MCP (Claude Desktop, Hermes, agentes personalizados) pueda conectarse e invocar las herramientas comerciales.
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"
# }
# }
# }
# }Ejemplo en 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())Ejemplo en 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();Salida esperada
JSON
Found 5 articles for NVDA
NVIDIA Q2 2026 Earnings Beat Expectations...
NVDA Stock Surges on New AI Chip Announcement...