如何构建结合搜索和金融数据的交易数据MCP服务器

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"]
            }
        )
    ]

@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))]

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))]

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"
#       }
#     }
#   }
# }