بهبود Claude Desktop با سرور MCP

# utils.py
import requests
import os
from dotenv import load_dotenv

load_dotenv()

ALPHA_VANTAGE_API_KEY = os.getenv("ALPHAVANTAGE_API_KEY")
BASE_URL = "https://www.alphavantage.co/query"

def get_stock_sentiment(symbol: str):
    """بازیابی احساسات خبری سهام از آلفا ونتیج."""
    params = {
        "function": "NEWS_SENTIMENT",
        "tickers": symbol,
        "apikey": ALPHA_VANTAGE_API_KEY,
        "sort": "latest"
    }
    try:
        response = requests.get(BASE_URL, params=params, timeout=10)
        response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
        data = response.json()
        news = data.get('feed', [])
        if news:
            top_news = news[:5]  # Get top 5 news items
sentiment_scores = [item['sentiment_score'] for item in top_news]
            average_sentiment = sum(float(score) for score in sentiment_scores) / len(sentiment_scores) if sentiment_scores else 0
            return {
                "symbol": symbol,
                "average_sentiment_score": average_sentiment,
                "news": [{
                    "title": item['title'],
                    "url": item['url'],
                    "source": item['source'],
                    "sentiment_score": item['sentiment_score']
                } for item in top_news]
            }
        else:
            return {"symbol": symbol, "message": "No news found"}
    except requests.exceptions.RequestException as e:
        print(f"API Request failed: {e}")
        return {"symbol": symbol, "message": f"API Request failed: {e}"}
    except (KeyError, ValueError) as e:
        print(f"Error processing API response: {e}")
        return {"symbol": symbol, "message": f"Error processing API response: {e}"}


def get_top_gainers():
    """بازیابی 10 سودآورترین روزانه."""
    params = {
        "function": "TOP_GAINERS_LOSERS",
        "apikey": ALPHA_VANTAGE_API_KEY
    }
    try:
        response = requests.get(BASE_URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        top_gainers = data.get('top_gainers', [])
        return [{
            "ticker": item['ticker'],
            "price": item['price'],
            "change_amount": item['change_amount'],
            "change_percentage": item['change_percentage'],
            "volume": item['volume']
        } for item in top_gainers]
    except requests.exceptions.RequestException as e:
        print(f"API Request failed: {e}")
        return {"message": f"API Request failed: {e}"}
    except (KeyError, ValueError) as e:
        print(f"Error processing API response: {e}")
        return {"message": f"Error processing API response: {e}"}

def get_top_losers():
    """بازیابی 10 بازنده برتر روزانه."""
    params = {
        "function": "TOP_GAINERS_LOSERS",
        "apikey": ALPHA_VANTAGE_API_KEY
    }
    try:
        response = requests.get(BASE_URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        top_losers = data.get('top_losers', [])
        return [{
            "ticker": item['ticker'],
            "price": item['price'],
            "change_amount": item['change_amount'],
            "change_percentage": item['change_percentage'],
            "volume": item['volume']
        } for item in top_losers]
    except requests.exceptions.RequestException as e:
        print(f"API Request failed: {e}")
        return {"message": f"API Request failed: {e}"}
    except (KeyError, ValueError) as e:
        print(f"Error processing API response: {e}")
        return {"message": f"Error processing API response: {e}"}

def get_most_active():
    """بازیابی 10 سهام پرطرفدار."""
    params = {
        "function": "TOP_GAINERS_LOSERS",
        "apikey": ALPHA_VANTAGE_API_KEY
    }
    try:
        response = requests.get(BASE_URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        most_active = data.get('most_active', [])
        return [{
            "ticker": item['ticker'],
            "price": item['price'],
            "change_amount": item['change_amount'],
            "change_percentage": item['change_percentage'],
            "volume": item['volume']
        } for item in most_active]
    except requests.exceptions.RequestException as e:
        print(f"API Request failed: {e}")
        return {"message": f"API Request failed: {e}"}
    except (KeyError, ValueError) as e:
        print(f"Error processing API response: {e}")
        return {"message": f"Error processing API response: {e}"}

# main.py
from fastapi import FastAPI, HTTPException
from utils import get_stock_sentiment, get_top_gainers, get_top_losers, get_most_active

app = FastAPI()

@app.get("/stock_sentiment/{symbol}")
async def stock_sentiment(symbol: str):
    """بازیابی احساسات خبری سهام برای یک نماد."""
    result = get_stock_sentiment(symbol)
    if "message" in result:
        raise HTTPException(status_code=404, detail=result["message"])
    return result

@app.get("/top_gainers")
async def top_gainers():
    """بازیابی 10 سودآورترین روزانه."""
    result = get_top_gainers()
    if "message" in result:
        raise HTTPException(status_code=500, detail=result["message"])
    return result

@app.get("/top_losers")
async def top_losers():
    """بازیابی 10 بازنده برتر روزانه."""
    result = get_top_losers()
    if "message" in result:
        raise HTTPException(status_code=500, detail=result["message"])
    return result

@app.get("/most_active")
async def most_active():
    """بازیابی 10 سهام پرطرفدار."""
    result = get_most_active()
    if "message" in result:
        raise HTTPException(status_code=500, detail=result["message"])
    return result

import logging
import requests
import os
from dotenv import load_dotenv
import time

load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

ALPHA_VANTAGE_API_KEY = os.getenv("ALPHAVANTAGE_API_KEY")
BASE_URL = "https://www.alphavantage.co/query"

def get_stock_sentiment(symbol: str, max_retries=3):
    """بازیابی احساسات خبری سهام از آلفا ونتیج با retry."""
    params = {
        "function": "NEWS_SENTIMENT",
        "tickers": symbol,
        "apikey": ALPHA_VANTAGE_API_KEY,
        "sort": "latest"
    }
    for retry in range(max_retries):
        try:
            response = requests.get(BASE_URL, params=params, timeout=10)
            response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
            data = response.json()
            news = data.get('feed', [])
            if news:
                top_news = news[:5]  # Get top 5 news items
                sentiment_scores = [item['sentiment_score'] for item in top_news]
                average_sentiment = sum(float(score) for score in sentiment_scores) / len(sentiment_scores) if sentiment_scores else 0
                return {
                    "symbol": symbol,
                    "average_sentiment_score": average_sentiment,
                    "news": [{
                        "title": item['title'],
                        "url": item['url'],
                        "source": item['source'],
                        "sentiment_score": item['sentiment_score']
                    } for item in top_news]
                }
            else:
                return {"symbol": symbol, "message": "No news found"}
        except requests.exceptions.RequestException as e:
            logging.error(f"API Request failed (Retry {retry + 1}/{max_retries}): {e}")
            if retry < max_retries - 1:
                time.sleep(2 ** retry)  # Exponential backoff
            else:
                return {"symbol": symbol, "message": f"API Request failed after multiple retries: {e}"}
        except (KeyError, ValueError) as e:
            logging.error(f"Error processing API response: {e}")
            return {"symbol": symbol, "message": f"Error processing API response: {e}"}
    return {"symbol": symbol, "message": "Failed to retrieve data after multiple retries."}

from functools import lru_cache

@lru_cache(maxsize=128)
def get_stock_sentiment(symbol: str):
    """بازیابی احساسات خبری سهام از آلفا ونتیج با caching."""
    # ... (کد API همانطور که قبلاً نشان داده شد)

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import APIKeyHeader

app = FastAPI()
API_KEY = "YOUR_API_KEY"  # Replace with a secure method to manage API keys
API_KEY_NAME = "X-API-Key"

api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True)

async def verify_api_key(api_key: str = Depends(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return api_key

@app.get("/stock_sentiment/{symbol}")
async def stock_sentiment(symbol: str, api_key: str = Depends(verify_api_key)):
    """بازیابی احساسات خبری سهام برای یک نماد."""
    result = get_stock_sentiment(symbol)
    if "message" in result:
        raise HTTPException(status_code=404, detail=result["message"])
    return result