Cómo crear un servidor MCP de noticias financieras con búsqueda

import os, requests, re, time
from datetime import datetime

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
URL = 'https://api.scavio.dev/api/v1/search'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

def search_stock_news(ticker: str, num: int = 5) -> list:
    """Get latest news for a stock ticker."""
    resp = requests.post(URL, headers=H,
        json={'query': f'{ticker} stock news 2026', 'country_code': 'us', 'num_results': num})
    return [{'title': r['title'], 'snippet': r.get('snippet', ''),
             'url': r['link'], 'source': 'google_news'}
            for r in resp.json().get('organic_results', [])]

def search_earnings(ticker: str) -> list:
    """Search for recent earnings reports."""
    resp = requests.post(URL, headers=H,
        json={'query': f'{ticker} earnings report Q2 2026',
              'country_code': 'us', 'num_results': 5})
    return [{'title': r['title'], 'snippet': r.get('snippet', ''),
             'url': r['link'], 'source': 'earnings'}
            for r in resp.json().get('organic_results', [])]

def search_market_sentiment(ticker: str) -> list:
    """Check Reddit for market sentiment."""
    resp = requests.post(URL, headers=H,
        json={'query': f'site:reddit.com {ticker} stock',
              'country_code': 'us', 'num_results': 5})
    return [{'title': r['title'], 'snippet': r.get('snippet', ''),
             'url': r['link'], 'source': 'reddit'}
            for r in resp.json().get('organic_results', [])]

news = search_stock_news('AAPL')
print(f'AAPL news: {len(news)} articles')
for n in news[:3]:
    print(f'  {n["title"][:60]}')

def stock_report(ticker: str) -> dict:
    """Generate a comprehensive stock report."""
    news = search_stock_news(ticker)
    time.sleep(0.3)
    earnings = search_earnings(ticker)
    time.sleep(0.3)
    sentiment = search_market_sentiment(ticker)
    # Analyze sentiment from Reddit
    all_reddit_text = ' '.join(r['snippet'] for r in sentiment).lower()
    bullish_words = ['bull', 'buy', 'moon', 'undervalued', 'growth', 'strong']
    bearish_words = ['bear', 'sell', 'overvalued', 'crash', 'decline', 'weak']
    bull_count = sum(1 for w in bullish_words if w in all_reddit_text)
    bear_count = sum(1 for w in bearish_words if w in all_reddit_text)
    sentiment_label = 'bullish' if bull_count > bear_count else 'bearish' if bear_count > bull_count else 'neutral'
    # Extract price mentions
    all_text = ' '.join(r['snippet'] for r in news + earnings)
    prices = re.findall(r'\$([\d,]+\.?\d*)', all_text)
    return {
        'ticker': ticker,
        'timestamp': datetime.now().isoformat(),
        'news_count': len(news),
        'top_headlines': [n['title'] for n in news[:3]],
        'earnings_count': len(earnings),
        'reddit_sentiment': sentiment_label,
        'sentiment_detail': {'bullish': bull_count, 'bearish': bear_count},
        'price_mentions': prices[:5],
        'sources': news + earnings + sentiment,
        'credits_used': 3,
        'cost': 0.015,
    }

report = stock_report('NVDA')
print(f"Stock Report: {report['ticker']}")
print(f"Headlines: {len(report['top_headlines'])}")
for h in report['top_headlines']:
    print(f'  - {h[:60]}')
print(f"Reddit sentiment: {report['reddit_sentiment']}")
print(f"Cost: ${report['cost']}")

MCP_TOOLS = {
    'stock_news': {
        'name': 'stock_news',
        'description': 'Get the latest news articles for a stock ticker symbol. Returns headlines, snippets, and source URLs.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'ticker': {'type': 'string', 'description': 'Stock ticker symbol (e.g., AAPL, NVDA, TSLA)'},
                'num_results': {'type': 'integer', 'description': 'Number of articles (1-10)', 'default': 5}
            },
            'required': ['ticker']
        }
    },
    'stock_report': {
        'name': 'stock_report',
        'description': 'Generate a comprehensive stock report with news, earnings, and Reddit sentiment analysis.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'ticker': {'type': 'string', 'description': 'Stock ticker symbol'}
            },
            'required': ['ticker']
        }
    },
    'market_sentiment': {
        'name': 'market_sentiment',
        'description': 'Check Reddit sentiment for a stock or market topic.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'ticker': {'type': 'string', 'description': 'Stock ticker or topic'}
            },
            'required': ['ticker']
        }
    }
}

def handle_mcp_call(tool_name: str, args: dict) -> str:
    if tool_name == 'stock_news':
        news = search_stock_news(args['ticker'], args.get('num_results', 5))
        return '\n\n'.join([f'{n["title"]}\n{n["snippet"]}\nSource: {n["url"]}' for n in news])
    elif tool_name == 'stock_report':
        report = stock_report(args['ticker'])
        lines = [f'Stock Report: {report["ticker"]}',
                 f'Sentiment: {report["reddit_sentiment"]}',
                 f'Headlines:']
        lines.extend(f'  - {h}' for h in report['top_headlines'])
        return '\n'.join(lines)
    elif tool_name == 'market_sentiment':
        results = search_market_sentiment(args['ticker'])
        return '\n\n'.join(f'{r["title"]}\n{r["snippet"]}' for r in results)
    return 'Unknown tool'

print('MCP tools defined:', list(MCP_TOOLS.keys()))
result = handle_mcp_call('stock_news', {'ticker': 'AAPL', 'num_results': 3})
print(result[:300])

import json

def create_finance_mcp_config(output_path: str = '.mcp.json'):
    """Create MCP config with Scavio for financial data."""
    config = {
        'mcpServers': {
            'scavio': {
                'url': 'https://mcp.scavio.dev/mcp',
                'headers': {
                    'Authorization': 'Bearer ${SCAVIO_API_KEY}'
                }
            }
        }
    }
    with open(output_path, 'w') as f:
        json.dump(config, f, indent=2)
    print(f'MCP config written to {output_path}')
    print(f'Financial tools available via Scavio MCP:')
    print(f'  - Web search (stock news, earnings, filings)')
    print(f'  - Reddit search (market sentiment)')
    print(f'  - YouTube search (analyst videos)')
    print(f'  Cost: $0.005 per search')
    print(f'  Endpoint: mcp.scavio.dev/mcp')

# Test the full pipeline
create_finance_mcp_config()
print('\nExample usage with Claude:')
print('  User: "What is the latest news on NVDA stock?"')
print('  Agent calls: stock_news({ticker: "NVDA"})')
print('  Agent gets: real-time news headlines + Reddit sentiment')
print('  Cost: $0.005-$0.015 per question')