Building a Financial Analyst Agent with MCP

Financial analysis requires synthesizing information from multiple sources: market data, company filings, economic indicators, news sentiment, and technical charts. Human analysts spend hours gathering and processing this data. AI agents can automate much of this work while maintaining the analytical rigor that finance demands.

This tutorial builds a financial analyst agent using CrewAI for orchestration, Ollama for local LLM inference, and Model Context Protocol (MCP) servers for structured data access. The result is an agent that can research investments, analyze financial metrics, and produce professional-quality reports.

System Architecture

Agent Roles

Our financial analyst system has specialized agents for different aspects of analysis:

1. Market Data Analyst Gathers and interprets market data, price movements, and trading patterns.

2. Fundamental Analyst Evaluates company financials, business models, and competitive position.

3. Technical Analyst Analyzes price charts, indicators, and trading signals.

4. Sentiment Analyst Monitors news, social media, and market sentiment.

5. Risk Analyst Assesses investment risks and portfolio implications.

6. Report Compiler Synthesizes analysis into actionable investment reports.

MCP Integration

MCP servers provide structured access to financial data:

┌─────────────────────────────────────────────┐
│             Financial Analyst Agent          │
├─────────────────────────────────────────────┤
│                  CrewAI                      │
│  ┌─────────┐  ┌──────────┐  ┌────────────┐ │
│  │ Market  │  │Fundamental│  │ Technical  │ │
│  │ Agent   │  │  Agent    │  │   Agent    │ │
│  └────┬────┘  └────┬─────┘  └─────┬──────┘ │
│       │            │              │         │
├───────┴────────────┴──────────────┴─────────┤
│                 MCP Layer                    │
│  ┌────────┐  ┌─────────┐  ┌──────────────┐ │
│  │ Yahoo  │  │  Alpha  │  │    News      │ │
│  │Finance │  │ Vantage │  │    API       │ │
│  │  MCP   │  │   MCP   │  │    MCP       │ │
│  └────────┘  └─────────┘  └──────────────┘ │
└─────────────────────────────────────────────┘

Project Setup

Project Structure

financial-analyst/
├── agents/
│   ├── __init__.py
│   ├── market_analyst.py
│   ├── fundamental_analyst.py
│   ├── technical_analyst.py
│   ├── sentiment_analyst.py
│   ├── risk_analyst.py
│   └── report_compiler.py
├── mcp_servers/
│   ├── __init__.py
│   ├── yahoo_finance.py
│   ├── alpha_vantage.py
│   └── news_api.py
├── tools/
│   ├── __init__.py
│   ├── financial_tools.py
│   └── chart_tools.py
├── workflows/
│   ├── __init__.py
│   ├── stock_analysis.py
│   └── portfolio_review.py
├── output/
│   └── (generated reports)
├── config/
│   └── settings.py
├── main.py
└── requirements.txt

Dependencies

# requirements.txt
crewai>=0.28.0
crewai-tools>=0.2.0
ollama>=0.1.0
mcp>=0.9.0
yfinance>=0.2.0
pandas>=2.0.0
numpy>=1.24.0
ta>=0.10.0  # Technical analysis library
plotly>=5.0.0
python-dotenv>=1.0.0
aiohttp>=3.9.0

Configuration

# config/settings.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # Ollama
    ollama_base_url: str = "http://localhost:11434"
    ollama_model: str = "llama3:8b"

    # API Keys
    alpha_vantage_api_key: Optional[str] = None
    news_api_key: Optional[str] = None

    # Analysis settings
    default_period: str = "1y"  # 1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y
    default_interval: str = "1d"

    # Risk parameters
    risk_free_rate: float = 0.05  # 5% annual
    confidence_level: float = 0.95

    class Config:
        env_file = ".env"

settings = Settings()

Building MCP Servers

Yahoo Finance MCP Server

# mcp_servers/yahoo_finance.py
import yfinance as yf
from mcp.server import Server
from mcp.types import Tool, TextContent
import json
from datetime import datetime, timedelta

server = Server("yahoo-finance")

@server.tool()
async def get_stock_info(symbol: str) -> str:
    """
    Get basic information about a stock.

    Args:
        symbol: Stock ticker symbol (e.g., AAPL, GOOGL)
    """
    ticker = yf.Ticker(symbol)
    info = ticker.info

    return json.dumps({
        "symbol": symbol,
        "name": info.get("longName", "Unknown"),
        "sector": info.get("sector", "Unknown"),
        "industry": info.get("industry", "Unknown"),
        "market_cap": info.get("marketCap"),
        "pe_ratio": info.get("trailingPE"),
        "forward_pe": info.get("forwardPE"),
        "dividend_yield": info.get("dividendYield"),
        "beta": info.get("beta"),
        "52_week_high": info.get("fiftyTwoWeekHigh"),
        "52_week_low": info.get("fiftyTwoWeekLow"),
        "current_price": info.get("currentPrice"),
        "target_price": info.get("targetMeanPrice"),
        "recommendation": info.get("recommendationKey")
    }, indent=2)

@server.tool()
async def get_price_history(
    symbol: str,
    period: str = "1y",
    interval: str = "1d"
) -> str:
    """
    Get historical price data for a stock.

    Args:
        symbol: Stock ticker symbol
        period: Data period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
        interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
    """
    ticker = yf.Ticker(symbol)
    hist = ticker.history(period=period, interval=interval)

    if hist.empty:
        return json.dumps({"error": f"No data found for {symbol}"})

    # Convert to list of records
    records = []
    for date, row in hist.iterrows():
        records.append({
            "date": date.strftime("%Y-%m-%d"),
            "open": round(row["Open"], 2),
            "high": round(row["High"], 2),
            "low": round(row["Low"], 2),
            "close": round(row["Close"], 2),
            "volume": int(row["Volume"])
        })

    return json.dumps({
        "symbol": symbol,
        "period": period,
        "interval": interval,
        "data_points": len(records),
        "prices": records[-30:]  # Last 30 data points
    }, indent=2)

@server.tool()
async def get_financials(symbol: str) -> str:
    """
    Get financial statements for a company.

    Args:
        symbol: Stock ticker symbol
    """
    ticker = yf.Ticker(symbol)

    # Get financial data
    income_stmt = ticker.income_stmt
    balance_sheet = ticker.balance_sheet
    cash_flow = ticker.cashflow

    result = {
        "symbol": symbol,
        "income_statement": {},
        "balance_sheet": {},
        "cash_flow": {}
    }

    # Extract key metrics from most recent year
    if not income_stmt.empty:
        latest = income_stmt.iloc[:, 0]
        result["income_statement"] = {
            "revenue": _safe_get(latest, "Total Revenue"),
            "gross_profit": _safe_get(latest, "Gross Profit"),
            "operating_income": _safe_get(latest, "Operating Income"),
            "net_income": _safe_get(latest, "Net Income"),
            "ebitda": _safe_get(latest, "EBITDA")
        }

    if not balance_sheet.empty:
        latest = balance_sheet.iloc[:, 0]
        result["balance_sheet"] = {
            "total_assets": _safe_get(latest, "Total Assets"),
            "total_liabilities": _safe_get(latest, "Total Liabilities Net Minority Interest"),
            "total_equity": _safe_get(latest, "Stockholders Equity"),
            "cash": _safe_get(latest, "Cash And Cash Equivalents"),
            "total_debt": _safe_get(latest, "Total Debt")
        }

    if not cash_flow.empty:
        latest = cash_flow.iloc[:, 0]
        result["cash_flow"] = {
            "operating_cash_flow": _safe_get(latest, "Operating Cash Flow"),
            "investing_cash_flow": _safe_get(latest, "Investing Cash Flow"),
            "financing_cash_flow": _safe_get(latest, "Financing Cash Flow"),
            "free_cash_flow": _safe_get(latest, "Free Cash Flow")
        }

    return json.dumps(result, indent=2)

def _safe_get(series, key):
    try:
        value = series.get(key)
        if value is not None:
            return float(value)
    except:
        pass
    return None

@server.tool()
async def compare_stocks(symbols: list[str]) -> str:
    """
    Compare multiple stocks on key metrics.

    Args:
        symbols: List of stock ticker symbols to compare
    """
    comparisons = []

    for symbol in symbols:
        ticker = yf.Ticker(symbol)
        info = ticker.info

        comparisons.append({
            "symbol": symbol,
            "name": info.get("longName", "Unknown"),
            "market_cap": info.get("marketCap"),
            "pe_ratio": info.get("trailingPE"),
            "price_to_book": info.get("priceToBook"),
            "profit_margin": info.get("profitMargins"),
            "revenue_growth": info.get("revenueGrowth"),
            "return_on_equity": info.get("returnOnEquity"),
            "debt_to_equity": info.get("debtToEquity"),
            "current_ratio": info.get("currentRatio"),
            "52_week_change": info.get("52WeekChange")
        })

    return json.dumps({
        "comparison_date": datetime.now().strftime("%Y-%m-%d"),
        "stocks": comparisons
    }, indent=2)

if __name__ == "__main__":
    import asyncio
    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)

    asyncio.run(main())

Technical Analysis Tools

# tools/financial_tools.py
from crewai_tools import tool
import yfinance as yf
import pandas as pd
import numpy as np
from ta import trend, momentum, volatility, volume
import json

class TechnicalAnalysisTools:
    @tool("Calculate Technical Indicators")
    def calculate_indicators(self, symbol: str, period: str = "1y") -> str:
        """
        Calculate technical indicators for a stock.

        Args:
            symbol: Stock ticker symbol
            period: Data period
        """
        ticker = yf.Ticker(symbol)
        df = ticker.history(period=period)

        if df.empty:
            return json.dumps({"error": f"No data for {symbol}"})

        # Calculate indicators
        indicators = {}

        # Moving Averages
        indicators["sma_20"] = round(df["Close"].rolling(20).mean().iloc[-1], 2)
        indicators["sma_50"] = round(df["Close"].rolling(50).mean().iloc[-1], 2)
        indicators["sma_200"] = round(df["Close"].rolling(200).mean().iloc[-1], 2)
        indicators["ema_12"] = round(df["Close"].ewm(span=12).mean().iloc[-1], 2)
        indicators["ema_26"] = round(df["Close"].ewm(span=26).mean().iloc[-1], 2)

        # RSI
        rsi = momentum.RSIIndicator(df["Close"])
        indicators["rsi"] = round(rsi.rsi().iloc[-1], 2)

        # MACD
        macd = trend.MACD(df["Close"])
        indicators["macd"] = round(macd.macd().iloc[-1], 4)
        indicators["macd_signal"] = round(macd.macd_signal().iloc[-1], 4)
        indicators["macd_histogram"] = round(macd.macd_diff().iloc[-1], 4)

        # Bollinger Bands
        bb = volatility.BollingerBands(df["Close"])
        indicators["bb_upper"] = round(bb.bollinger_hband().iloc[-1], 2)
        indicators["bb_middle"] = round(bb.bollinger_mavg().iloc[-1], 2)
        indicators["bb_lower"] = round(bb.bollinger_lband().iloc[-1], 2)

        # ATR (Average True Range)
        atr = volatility.AverageTrueRange(df["High"], df["Low"], df["Close"])
        indicators["atr"] = round(atr.average_true_range().iloc[-1], 2)

        # Volume indicators
        indicators["volume_sma_20"] = int(df["Volume"].rolling(20).mean().iloc[-1])
        indicators["current_volume"] = int(df["Volume"].iloc[-1])

        # Price context
        indicators["current_price"] = round(df["Close"].iloc[-1], 2)
        indicators["price_change_1d"] = round(
            (df["Close"].iloc[-1] - df["Close"].iloc[-2]) / df["Close"].iloc[-2] * 100, 2
        )

        return json.dumps({
            "symbol": symbol,
            "date": df.index[-1].strftime("%Y-%m-%d"),
            "indicators": indicators
        }, indent=2)

    @tool("Identify Chart Patterns")
    def identify_patterns(self, symbol: str, period: str = "6mo") -> str:
        """
        Identify chart patterns in price data.

        Args:
            symbol: Stock ticker symbol
            period: Data period to analyze
        """
        ticker = yf.Ticker(symbol)
        df = ticker.history(period=period)

        if df.empty:
            return json.dumps({"error": f"No data for {symbol}"})

        patterns = []
        closes = df["Close"].values
        highs = df["High"].values
        lows = df["Low"].values

        # Trend identification
        sma_20 = df["Close"].rolling(20).mean()
        sma_50 = df["Close"].rolling(50).mean()

        if sma_20.iloc[-1] > sma_50.iloc[-1]:
            if sma_20.iloc[-20] < sma_50.iloc[-20]:
                patterns.append({
                    "pattern": "Golden Cross",
                    "type": "bullish",
                    "description": "Short-term MA crossed above long-term MA"
                })
            else:
                patterns.append({
                    "pattern": "Uptrend",
                    "type": "bullish",
                    "description": "Price above moving averages"
                })
        else:
            if sma_20.iloc[-20] > sma_50.iloc[-20]:
                patterns.append({
                    "pattern": "Death Cross",
                    "type": "bearish",
                    "description": "Short-term MA crossed below long-term MA"
                })
            else:
                patterns.append({
                    "pattern": "Downtrend",
                    "type": "bearish",
                    "description": "Price below moving averages"
                })

        # Support/Resistance levels
        recent_high = max(highs[-20:])
        recent_low = min(lows[-20:])
        current = closes[-1]

        patterns.append({
            "pattern": "Support/Resistance",
            "type": "neutral",
            "resistance": round(recent_high, 2),
            "support": round(recent_low, 2),
            "current": round(current, 2),
            "position": "near_resistance" if current > (recent_high * 0.95) else
                       "near_support" if current < (recent_low * 1.05) else "middle"
        })

        # RSI divergence check
        rsi = momentum.RSIIndicator(df["Close"])
        rsi_values = rsi.rsi().values

        if rsi_values[-1] > 70:
            patterns.append({
                "pattern": "RSI Overbought",
                "type": "bearish",
                "rsi": round(rsi_values[-1], 2)
            })
        elif rsi_values[-1] < 30:
            patterns.append({
                "pattern": "RSI Oversold",
                "type": "bullish",
                "rsi": round(rsi_values[-1], 2)
            })

        return json.dumps({
            "symbol": symbol,
            "patterns": patterns
        }, indent=2)

    @tool("Calculate Risk Metrics")
    def calculate_risk_metrics(
        self,
        symbol: str,
        benchmark: str = "SPY",
        period: str = "1y"
    ) -> str:
        """
        Calculate risk metrics for a stock.

        Args:
            symbol: Stock ticker symbol
            benchmark: Benchmark symbol for comparison
            period: Analysis period
        """
        from config.settings import settings

        # Get data
        stock = yf.Ticker(symbol)
        bench = yf.Ticker(benchmark)

        stock_hist = stock.history(period=period)
        bench_hist = bench.history(period=period)

        if stock_hist.empty or bench_hist.empty:
            return json.dumps({"error": "Could not fetch data"})

        # Calculate returns
        stock_returns = stock_hist["Close"].pct_change().dropna()
        bench_returns = bench_hist["Close"].pct_change().dropna()

        # Align dates
        common_dates = stock_returns.index.intersection(bench_returns.index)
        stock_returns = stock_returns[common_dates]
        bench_returns = bench_returns[common_dates]

        metrics = {}

        # Volatility (annualized)
        metrics["volatility"] = round(stock_returns.std() * np.sqrt(252) * 100, 2)

        # Beta
        covariance = np.cov(stock_returns, bench_returns)[0][1]
        bench_variance = bench_returns.var()
        metrics["beta"] = round(covariance / bench_variance, 3)

        # Alpha (annualized)
        stock_annual_return = (1 + stock_returns.mean()) ** 252 - 1
        bench_annual_return = (1 + bench_returns.mean()) ** 252 - 1
        expected_return = settings.risk_free_rate + metrics["beta"] * (bench_annual_return - settings.risk_free_rate)
        metrics["alpha"] = round((stock_annual_return - expected_return) * 100, 2)

        # Sharpe Ratio
        excess_return = stock_annual_return - settings.risk_free_rate
        metrics["sharpe_ratio"] = round(excess_return / (stock_returns.std() * np.sqrt(252)), 3)

        # Sortino Ratio
        downside_returns = stock_returns[stock_returns < 0]
        downside_std = downside_returns.std() * np.sqrt(252)
        metrics["sortino_ratio"] = round(excess_return / downside_std, 3) if downside_std > 0 else None

        # Maximum Drawdown
        cumulative = (1 + stock_returns).cumprod()
        rolling_max = cumulative.expanding().max()
        drawdowns = (cumulative - rolling_max) / rolling_max
        metrics["max_drawdown"] = round(drawdowns.min() * 100, 2)

        # Value at Risk (95%)
        metrics["var_95"] = round(np.percentile(stock_returns, 5) * 100, 2)

        return json.dumps({
            "symbol": symbol,
            "benchmark": benchmark,
            "period": period,
            "metrics": metrics
        }, indent=2)

Building the Agents

Market Data Analyst

# agents/market_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_market_analyst(llm: Ollama, mcp_tools: list) -> Agent:
    return Agent(
        role='Market Data Analyst',
        goal='Gather and interpret market data, price movements, and trading patterns',
        backstory='''You are a quantitative market analyst with deep expertise in
        financial markets. You excel at:
        - Analyzing price and volume data
        - Identifying market trends and patterns
        - Interpreting market indicators
        - Understanding market microstructure
        - Contextualizing movements with market events''',
        llm=llm,
        tools=mcp_tools,
        verbose=True
    )

def create_market_analysis_task(agent: Agent, symbol: str):
    return f"""Analyze the current market data for {symbol}:

1. **Price Analysis**
   - Current price and recent changes
   - Position relative to 52-week range
   - Comparison to key moving averages

2. **Volume Analysis**
   - Current volume vs. average
   - Volume trends
   - Any unusual activity

3. **Market Context**
   - How is the broader market performing?
   - Sector performance
   - Any relevant market events

4. **Key Metrics**
   - Market cap and valuation multiples
   - Volatility measures
   - Liquidity indicators

Provide factual data with brief interpretation."""

Fundamental Analyst

# agents/fundamental_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_fundamental_analyst(llm: Ollama, mcp_tools: list) -> Agent:
    return Agent(
        role='Fundamental Analyst',
        goal='Evaluate company financials, business model, and intrinsic value',
        backstory='''You are a senior fundamental analyst with expertise in
        financial statement analysis and valuation. You:
        - Analyze income statements, balance sheets, and cash flows
        - Calculate and interpret financial ratios
        - Evaluate competitive positioning
        - Assess management quality and strategy
        - Estimate intrinsic value using multiple methods''',
        llm=llm,
        tools=mcp_tools,
        verbose=True
    )

def create_fundamental_analysis_task(agent: Agent, symbol: str):
    return f"""Perform fundamental analysis on {symbol}:

1. **Financial Health**
   - Revenue and earnings trends
   - Profit margins (gross, operating, net)
   - Return on equity and assets
   - Debt levels and coverage ratios

2. **Cash Flow Analysis**
   - Operating cash flow quality
   - Free cash flow generation
   - Capital allocation priorities

3. **Valuation**
   - P/E ratio vs. historical and peers
   - Price-to-book and price-to-sales
   - EV/EBITDA comparison
   - Dividend yield and payout ratio

4. **Business Quality**
   - Competitive advantages
   - Growth drivers
   - Key risks

5. **Fair Value Estimate**
   - Based on your analysis, is the stock:
     - Undervalued
     - Fairly valued
     - Overvalued
   - Provide reasoning"""

Technical Analyst

# agents/technical_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama
from tools.financial_tools import TechnicalAnalysisTools

def create_technical_analyst(llm: Ollama, tools: TechnicalAnalysisTools) -> Agent:
    return Agent(
        role='Technical Analyst',
        goal='Analyze price charts, indicators, and trading patterns',
        backstory='''You are an expert technical analyst who reads charts
        to identify trading opportunities. You:
        - Analyze price patterns and trends
        - Use technical indicators effectively
        - Identify support and resistance levels
        - Time entry and exit points
        - Manage risk with technical tools''',
        llm=llm,
        tools=[
            tools.calculate_indicators,
            tools.identify_patterns,
            tools.calculate_risk_metrics
        ],
        verbose=True
    )

def create_technical_analysis_task(agent: Agent, symbol: str):
    return f"""Perform technical analysis on {symbol}:

1. **Trend Analysis**
   - Primary trend direction
   - Moving average alignment
   - Trend strength

2. **Momentum**
   - RSI reading and interpretation
   - MACD signal
   - Momentum divergences

3. **Volatility**
   - Bollinger Band position
   - ATR for risk sizing
   - Volatility trend

4. **Chart Patterns**
   - Any recognizable patterns
   - Support/resistance levels
   - Breakout potential

5. **Trading Signals**
   - Current technical stance (bullish/bearish/neutral)
   - Key levels to watch
   - Risk/reward assessment"""

Risk Analyst

# agents/risk_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_risk_analyst(llm: Ollama, tools: list) -> Agent:
    return Agent(
        role='Risk Analyst',
        goal='Assess investment risks and portfolio implications',
        backstory='''You are a risk management specialist focused on
        protecting capital while capturing returns. You:
        - Quantify various risk factors
        - Stress test investment scenarios
        - Consider tail risks and black swans
        - Evaluate portfolio impact
        - Recommend position sizing''',
        llm=llm,
        tools=tools,
        verbose=True
    )

def create_risk_analysis_task(agent: Agent, symbol: str):
    return f"""Analyze investment risks for {symbol}:

1. **Market Risk**
   - Beta and correlation to market
   - Volatility assessment
   - Value at Risk

2. **Company-Specific Risk**
   - Business model risks
   - Financial leverage
   - Management/governance risks

3. **External Risks**
   - Regulatory risks
   - Competitive threats
   - Macroeconomic sensitivity

4. **Downside Scenarios**
   - What could go wrong?
   - Maximum drawdown potential
   - Stress test results

5. **Risk-Adjusted View**
   - Is the potential return worth the risk?
   - Suggested position sizing
   - Hedging considerations"""

Report Compiler

# agents/report_compiler.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_report_compiler(llm: Ollama) -> Agent:
    return Agent(
        role='Investment Report Compiler',
        goal='Synthesize analysis into actionable investment reports',
        backstory='''You are a senior investment strategist who creates
        compelling, actionable research reports. You:
        - Synthesize diverse analytical perspectives
        - Identify the most important factors
        - Communicate clearly for decision-making
        - Provide specific, actionable recommendations
        - Balance conviction with appropriate caveats''',
        llm=llm,
        verbose=True
    )

def create_report_task(agent: Agent, symbol: str):
    return f"""Compile a comprehensive investment report for {symbol}:

## Report Structure

### Executive Summary
- Investment thesis in 2-3 sentences
- Key recommendation (Buy/Hold/Sell)
- Target price if applicable
- Key risks

### Company Overview
- Brief business description
- Competitive position
- Recent developments

### Investment Analysis

#### Fundamental View
(Summarize fundamental analysis)

#### Technical View
(Summarize technical analysis)

#### Risk Assessment
(Summarize key risks)

### Recommendation

**Rating:** [Strong Buy / Buy / Hold / Sell / Strong Sell]

**Conviction Level:** [High / Medium / Low]

**Investment Horizon:** [Short-term / Medium-term / Long-term]

**Key Catalysts:**
- Near-term catalysts
- Long-term value drivers

**Risks to Thesis:**
- Primary risks
- Mitigating factors

### Action Items
- Specific entry/exit recommendations
- Position sizing guidance
- Monitoring points"""

Complete Workflow

Stock Analysis Crew

# workflows/stock_analysis.py
from crewai import Crew, Process, Task
from langchain_community.llms import Ollama
from mcp import ClientSession
from agents.market_analyst import create_market_analyst, create_market_analysis_task
from agents.fundamental_analyst import create_fundamental_analyst, create_fundamental_analysis_task
from agents.technical_analyst import create_technical_analyst, create_technical_analysis_task
from agents.risk_analyst import create_risk_analyst, create_risk_analysis_task
from agents.report_compiler import create_report_compiler, create_report_task
from tools.financial_tools import TechnicalAnalysisTools
from config.settings import settings
import asyncio

class StockAnalyzer:
    def __init__(self):
        # Initialize LLM
        self.llm = Ollama(
            model=settings.ollama_model,
            base_url=settings.ollama_base_url
        )

        # Initialize tools
        self.tech_tools = TechnicalAnalysisTools()

        # MCP tools will be loaded from servers
        self.mcp_tools = []

    async def connect_mcp_servers(self):
        """Connect to MCP servers for financial data."""
        # In production, this would connect to actual MCP servers
        # For now, we use the direct yfinance tools
        pass

    def analyze_stock(self, symbol: str) -> str:
        """Run full stock analysis."""

        # Create agents
        market_analyst = create_market_analyst(self.llm, self.mcp_tools)
        fundamental_analyst = create_fundamental_analyst(self.llm, self.mcp_tools)
        technical_analyst = create_technical_analyst(self.llm, self.tech_tools)
        risk_analyst = create_risk_analyst(self.llm, [self.tech_tools.calculate_risk_metrics])
        report_compiler = create_report_compiler(self.llm)

        # Create tasks
        market_task = Task(
            description=create_market_analysis_task(market_analyst, symbol),
            expected_output="Market data analysis",
            agent=market_analyst
        )

        fundamental_task = Task(
            description=create_fundamental_analysis_task(fundamental_analyst, symbol),
            expected_output="Fundamental analysis",
            agent=fundamental_analyst
        )

        technical_task = Task(
            description=create_technical_analysis_task(technical_analyst, symbol),
            expected_output="Technical analysis",
            agent=technical_analyst
        )

        risk_task = Task(
            description=create_risk_analysis_task(risk_analyst, symbol),
            expected_output="Risk analysis",
            agent=risk_analyst,
            context=[market_task, fundamental_task, technical_task]
        )

        report_task = Task(
            description=create_report_task(report_compiler, symbol),
            expected_output="Investment report",
            agent=report_compiler,
            context=[market_task, fundamental_task, technical_task, risk_task]
        )

        # Create and run crew
        crew = Crew(
            agents=[
                market_analyst,
                fundamental_analyst,
                technical_analyst,
                risk_analyst,
                report_compiler
            ],
            tasks=[
                market_task,
                fundamental_task,
                technical_task,
                risk_task,
                report_task
            ],
            process=Process.sequential,
            verbose=True
        )

        result = crew.kickoff()
        return str(result)

    def quick_analysis(self, symbol: str) -> str:
        """Quick analysis focusing on key metrics."""

        # Simplified analysis with fewer agents
        technical_analyst = create_technical_analyst(self.llm, self.tech_tools)
        report_compiler = create_report_compiler(self.llm)

        quick_analysis_task = Task(
            description=f"""Quick analysis of {symbol}:
            1. Calculate key technical indicators
            2. Identify current trend
            3. Note key support/resistance levels
            4. Provide a brief trading view""",
            expected_output="Quick technical overview",
            agent=technical_analyst
        )

        brief_report_task = Task(
            description=f"""Create a brief report for {symbol}:
            - Current technical stance
            - Key levels
            - Near-term outlook
            Keep it under 300 words.""",
            expected_output="Brief report",
            agent=report_compiler,
            context=[quick_analysis_task]
        )

        crew = Crew(
            agents=[technical_analyst, report_compiler],
            tasks=[quick_analysis_task, brief_report_task],
            process=Process.sequential,
            verbose=True
        )

        return str(crew.kickoff())

    def compare_stocks(self, symbols: list[str]) -> str:
        """Compare multiple stocks."""

        market_analyst = create_market_analyst(self.llm, self.mcp_tools)
        report_compiler = create_report_compiler(self.llm)

        comparison_task = Task(
            description=f"""Compare these stocks: {', '.join(symbols)}

            For each stock, gather:
            1. Current price and valuation
            2. Key financial metrics
            3. Technical position

            Then compare:
            - Which is cheapest by valuation?
            - Which has strongest momentum?
            - Which has best risk/reward?""",
            expected_output="Stock comparison data",
            agent=market_analyst
        )

        ranking_task = Task(
            description=f"""Based on the comparison, rank {', '.join(symbols)}:
            1. Best overall investment
            2. Best value play
            3. Best momentum play
            4. Most risky

            Explain your rankings.""",
            expected_output="Ranked comparison report",
            agent=report_compiler,
            context=[comparison_task]
        )

        crew = Crew(
            agents=[market_analyst, report_compiler],
            tasks=[comparison_task, ranking_task],
            process=Process.sequential,
            verbose=True
        )

        return str(crew.kickoff())

Main Entry Point

# main.py
from workflows.stock_analysis import StockAnalyzer
from datetime import datetime
import argparse

def main():
    parser = argparse.ArgumentParser(description="Financial Analyst Agent")
    parser.add_argument("symbol", help="Stock symbol to analyze")
    parser.add_argument(
        "--mode",
        choices=["full", "quick", "compare"],
        default="full",
        help="Analysis mode"
    )
    parser.add_argument(
        "--compare-with",
        nargs="+",
        help="Additional symbols for comparison mode"
    )
    parser.add_argument("--output", help="Output file path")

    args = parser.parse_args()

    analyzer = StockAnalyzer()

    print(f"Analyzing {args.symbol}...")
    print("-" * 50)

    if args.mode == "full":
        result = analyzer.analyze_stock(args.symbol)
    elif args.mode == "quick":
        result = analyzer.quick_analysis(args.symbol)
    elif args.mode == "compare":
        symbols = [args.symbol] + (args.compare_with or [])
        result = analyzer.compare_stocks(symbols)

    # Save output
    if args.output:
        output_path = args.output
    else:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_path = f"output/{args.symbol}_{args.mode}_{timestamp}.md"

    with open(output_path, "w") as f:
        f.write(result)

    print(f"\nAnalysis complete! Report saved to: {output_path}")

if __name__ == "__main__":
    main()

Usage Examples

# Full analysis
python main.py AAPL --mode full

# Quick analysis
python main.py MSFT --mode quick

# Compare stocks
python main.py AAPL --mode compare --compare-with MSFT GOOGL AMZN

# Specify output
python main.py TSLA --output reports/tesla_analysis.md

Summary

Building a financial analyst agent demonstrates how multi-agent systems can tackle complex, multi-faceted analysis tasks. By combining specialized agents with MCP-powered data access, you create a system that:

Gathers comprehensive data: Market prices, financials, technical indicators
Analyzes from multiple angles: Fundamental, technical, and risk perspectives
Synthesizes actionable insights: Clear recommendations with supporting analysis
Runs locally: Using Ollama for cost-effective inference

Key implementation points:

Specialize agents by analysis type: Each perspective gets dedicated focus
Use MCP for data access: Structured, reliable financial data retrieval
Layer analysis logically: Market data, then fundamental, then technical, then risk
Produce actionable output: Reports should support decision-making

This architecture extends naturally to portfolio analysis, market screening, and automated monitoring. The multi-agent foundation makes these extensions straightforward to implement.

This concludes the agent tutorials. Ready to learn about AI skills? Continue to What is an AI Skill? to understand how skills complement agents in the AI development toolkit.

Building a Financial Analyst Agent with MCP

System Architecture

Agent Roles

Our financial analyst system has specialized agents for different aspects of analysis:

1. Market Data Analyst Gathers and interprets market data, price movements, and trading patterns.

2. Fundamental Analyst Evaluates company financials, business models, and competitive position.

3. Technical Analyst Analyzes price charts, indicators, and trading signals.

4. Sentiment Analyst Monitors news, social media, and market sentiment.

5. Risk Analyst Assesses investment risks and portfolio implications.

6. Report Compiler Synthesizes analysis into actionable investment reports.

MCP Integration

MCP servers provide structured access to financial data:

┌─────────────────────────────────────────────┐
│             Financial Analyst Agent          │
├─────────────────────────────────────────────┤
│                  CrewAI                      │
│  ┌─────────┐  ┌──────────┐  ┌────────────┐ │
│  │ Market  │  │Fundamental│  │ Technical  │ │
│  │ Agent   │  │  Agent    │  │   Agent    │ │
│  └────┬────┘  └────┬─────┘  └─────┬──────┘ │
│       │            │              │         │
├───────┴────────────┴──────────────┴─────────┤
│                 MCP Layer                    │
│  ┌────────┐  ┌─────────┐  ┌──────────────┐ │
│  │ Yahoo  │  │  Alpha  │  │    News      │ │
│  │Finance │  │ Vantage │  │    API       │ │
│  │  MCP   │  │   MCP   │  │    MCP       │ │
│  └────────┘  └─────────┘  └──────────────┘ │
└─────────────────────────────────────────────┘

Project Setup

Project Structure

financial-analyst/
├── agents/
│   ├── __init__.py
│   ├── market_analyst.py
│   ├── fundamental_analyst.py
│   ├── technical_analyst.py
│   ├── sentiment_analyst.py
│   ├── risk_analyst.py
│   └── report_compiler.py
├── mcp_servers/
│   ├── __init__.py
│   ├── yahoo_finance.py
│   ├── alpha_vantage.py
│   └── news_api.py
├── tools/
│   ├── __init__.py
│   ├── financial_tools.py
│   └── chart_tools.py
├── workflows/
│   ├── __init__.py
│   ├── stock_analysis.py
│   └── portfolio_review.py
├── output/
│   └── (generated reports)
├── config/
│   └── settings.py
├── main.py
└── requirements.txt

Dependencies

# requirements.txt
crewai>=0.28.0
crewai-tools>=0.2.0
ollama>=0.1.0
mcp>=0.9.0
yfinance>=0.2.0
pandas>=2.0.0
numpy>=1.24.0
ta>=0.10.0  # Technical analysis library
plotly>=5.0.0
python-dotenv>=1.0.0
aiohttp>=3.9.0

Configuration

# config/settings.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # Ollama
    ollama_base_url: str = "http://localhost:11434"
    ollama_model: str = "llama3:8b"

    # API Keys
    alpha_vantage_api_key: Optional[str] = None
    news_api_key: Optional[str] = None

    # Analysis settings
    default_period: str = "1y"  # 1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y
    default_interval: str = "1d"

    # Risk parameters
    risk_free_rate: float = 0.05  # 5% annual
    confidence_level: float = 0.95

    class Config:
        env_file = ".env"

settings = Settings()

Building MCP Servers

Yahoo Finance MCP Server

# mcp_servers/yahoo_finance.py
import yfinance as yf
from mcp.server import Server
from mcp.types import Tool, TextContent
import json
from datetime import datetime, timedelta

server = Server("yahoo-finance")

@server.tool()
async def get_stock_info(symbol: str) -> str:
    """
    Get basic information about a stock.

    Args:
        symbol: Stock ticker symbol (e.g., AAPL, GOOGL)
    """
    ticker = yf.Ticker(symbol)
    info = ticker.info

    return json.dumps({
        "symbol": symbol,
        "name": info.get("longName", "Unknown"),
        "sector": info.get("sector", "Unknown"),
        "industry": info.get("industry", "Unknown"),
        "market_cap": info.get("marketCap"),
        "pe_ratio": info.get("trailingPE"),
        "forward_pe": info.get("forwardPE"),
        "dividend_yield": info.get("dividendYield"),
        "beta": info.get("beta"),
        "52_week_high": info.get("fiftyTwoWeekHigh"),
        "52_week_low": info.get("fiftyTwoWeekLow"),
        "current_price": info.get("currentPrice"),
        "target_price": info.get("targetMeanPrice"),
        "recommendation": info.get("recommendationKey")
    }, indent=2)

@server.tool()
async def get_price_history(
    symbol: str,
    period: str = "1y",
    interval: str = "1d"
) -> str:
    """
    Get historical price data for a stock.

    Args:
        symbol: Stock ticker symbol
        period: Data period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
        interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
    """
    ticker = yf.Ticker(symbol)
    hist = ticker.history(period=period, interval=interval)

    if hist.empty:
        return json.dumps({"error": f"No data found for {symbol}"})

    # Convert to list of records
    records = []
    for date, row in hist.iterrows():
        records.append({
            "date": date.strftime("%Y-%m-%d"),
            "open": round(row["Open"], 2),
            "high": round(row["High"], 2),
            "low": round(row["Low"], 2),
            "close": round(row["Close"], 2),
            "volume": int(row["Volume"])
        })

    return json.dumps({
        "symbol": symbol,
        "period": period,
        "interval": interval,
        "data_points": len(records),
        "prices": records[-30:]  # Last 30 data points
    }, indent=2)

@server.tool()
async def get_financials(symbol: str) -> str:
    """
    Get financial statements for a company.

    Args:
        symbol: Stock ticker symbol
    """
    ticker = yf.Ticker(symbol)

    # Get financial data
    income_stmt = ticker.income_stmt
    balance_sheet = ticker.balance_sheet
    cash_flow = ticker.cashflow

    result = {
        "symbol": symbol,
        "income_statement": {},
        "balance_sheet": {},
        "cash_flow": {}
    }

    # Extract key metrics from most recent year
    if not income_stmt.empty:
        latest = income_stmt.iloc[:, 0]
        result["income_statement"] = {
            "revenue": _safe_get(latest, "Total Revenue"),
            "gross_profit": _safe_get(latest, "Gross Profit"),
            "operating_income": _safe_get(latest, "Operating Income"),
            "net_income": _safe_get(latest, "Net Income"),
            "ebitda": _safe_get(latest, "EBITDA")
        }

    if not balance_sheet.empty:
        latest = balance_sheet.iloc[:, 0]
        result["balance_sheet"] = {
            "total_assets": _safe_get(latest, "Total Assets"),
            "total_liabilities": _safe_get(latest, "Total Liabilities Net Minority Interest"),
            "total_equity": _safe_get(latest, "Stockholders Equity"),
            "cash": _safe_get(latest, "Cash And Cash Equivalents"),
            "total_debt": _safe_get(latest, "Total Debt")
        }

    if not cash_flow.empty:
        latest = cash_flow.iloc[:, 0]
        result["cash_flow"] = {
            "operating_cash_flow": _safe_get(latest, "Operating Cash Flow"),
            "investing_cash_flow": _safe_get(latest, "Investing Cash Flow"),
            "financing_cash_flow": _safe_get(latest, "Financing Cash Flow"),
            "free_cash_flow": _safe_get(latest, "Free Cash Flow")
        }

    return json.dumps(result, indent=2)

def _safe_get(series, key):
    try:
        value = series.get(key)
        if value is not None:
            return float(value)
    except:
        pass
    return None

@server.tool()
async def compare_stocks(symbols: list[str]) -> str:
    """
    Compare multiple stocks on key metrics.

    Args:
        symbols: List of stock ticker symbols to compare
    """
    comparisons = []

    for symbol in symbols:
        ticker = yf.Ticker(symbol)
        info = ticker.info

        comparisons.append({
            "symbol": symbol,
            "name": info.get("longName", "Unknown"),
            "market_cap": info.get("marketCap"),
            "pe_ratio": info.get("trailingPE"),
            "price_to_book": info.get("priceToBook"),
            "profit_margin": info.get("profitMargins"),
            "revenue_growth": info.get("revenueGrowth"),
            "return_on_equity": info.get("returnOnEquity"),
            "debt_to_equity": info.get("debtToEquity"),
            "current_ratio": info.get("currentRatio"),
            "52_week_change": info.get("52WeekChange")
        })

    return json.dumps({
        "comparison_date": datetime.now().strftime("%Y-%m-%d"),
        "stocks": comparisons
    }, indent=2)

if __name__ == "__main__":
    import asyncio
    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)

    asyncio.run(main())

Technical Analysis Tools

# tools/financial_tools.py
from crewai_tools import tool
import yfinance as yf
import pandas as pd
import numpy as np
from ta import trend, momentum, volatility, volume
import json

class TechnicalAnalysisTools:
    @tool("Calculate Technical Indicators")
    def calculate_indicators(self, symbol: str, period: str = "1y") -> str:
        """
        Calculate technical indicators for a stock.

        Args:
            symbol: Stock ticker symbol
            period: Data period
        """
        ticker = yf.Ticker(symbol)
        df = ticker.history(period=period)

        if df.empty:
            return json.dumps({"error": f"No data for {symbol}"})

        # Calculate indicators
        indicators = {}

        # Moving Averages
        indicators["sma_20"] = round(df["Close"].rolling(20).mean().iloc[-1], 2)
        indicators["sma_50"] = round(df["Close"].rolling(50).mean().iloc[-1], 2)
        indicators["sma_200"] = round(df["Close"].rolling(200).mean().iloc[-1], 2)
        indicators["ema_12"] = round(df["Close"].ewm(span=12).mean().iloc[-1], 2)
        indicators["ema_26"] = round(df["Close"].ewm(span=26).mean().iloc[-1], 2)

        # RSI
        rsi = momentum.RSIIndicator(df["Close"])
        indicators["rsi"] = round(rsi.rsi().iloc[-1], 2)

        # MACD
        macd = trend.MACD(df["Close"])
        indicators["macd"] = round(macd.macd().iloc[-1], 4)
        indicators["macd_signal"] = round(macd.macd_signal().iloc[-1], 4)
        indicators["macd_histogram"] = round(macd.macd_diff().iloc[-1], 4)

        # Bollinger Bands
        bb = volatility.BollingerBands(df["Close"])
        indicators["bb_upper"] = round(bb.bollinger_hband().iloc[-1], 2)
        indicators["bb_middle"] = round(bb.bollinger_mavg().iloc[-1], 2)
        indicators["bb_lower"] = round(bb.bollinger_lband().iloc[-1], 2)

        # ATR (Average True Range)
        atr = volatility.AverageTrueRange(df["High"], df["Low"], df["Close"])
        indicators["atr"] = round(atr.average_true_range().iloc[-1], 2)

        # Volume indicators
        indicators["volume_sma_20"] = int(df["Volume"].rolling(20).mean().iloc[-1])
        indicators["current_volume"] = int(df["Volume"].iloc[-1])

        # Price context
        indicators["current_price"] = round(df["Close"].iloc[-1], 2)
        indicators["price_change_1d"] = round(
            (df["Close"].iloc[-1] - df["Close"].iloc[-2]) / df["Close"].iloc[-2] * 100, 2
        )

        return json.dumps({
            "symbol": symbol,
            "date": df.index[-1].strftime("%Y-%m-%d"),
            "indicators": indicators
        }, indent=2)

    @tool("Identify Chart Patterns")
    def identify_patterns(self, symbol: str, period: str = "6mo") -> str:
        """
        Identify chart patterns in price data.

        Args:
            symbol: Stock ticker symbol
            period: Data period to analyze
        """
        ticker = yf.Ticker(symbol)
        df = ticker.history(period=period)

        if df.empty:
            return json.dumps({"error": f"No data for {symbol}"})

        patterns = []
        closes = df["Close"].values
        highs = df["High"].values
        lows = df["Low"].values

        # Trend identification
        sma_20 = df["Close"].rolling(20).mean()
        sma_50 = df["Close"].rolling(50).mean()

        if sma_20.iloc[-1] > sma_50.iloc[-1]:
            if sma_20.iloc[-20] < sma_50.iloc[-20]:
                patterns.append({
                    "pattern": "Golden Cross",
                    "type": "bullish",
                    "description": "Short-term MA crossed above long-term MA"
                })
            else:
                patterns.append({
                    "pattern": "Uptrend",
                    "type": "bullish",
                    "description": "Price above moving averages"
                })
        else:
            if sma_20.iloc[-20] > sma_50.iloc[-20]:
                patterns.append({
                    "pattern": "Death Cross",
                    "type": "bearish",
                    "description": "Short-term MA crossed below long-term MA"
                })
            else:
                patterns.append({
                    "pattern": "Downtrend",
                    "type": "bearish",
                    "description": "Price below moving averages"
                })

        # Support/Resistance levels
        recent_high = max(highs[-20:])
        recent_low = min(lows[-20:])
        current = closes[-1]

        patterns.append({
            "pattern": "Support/Resistance",
            "type": "neutral",
            "resistance": round(recent_high, 2),
            "support": round(recent_low, 2),
            "current": round(current, 2),
            "position": "near_resistance" if current > (recent_high * 0.95) else
                       "near_support" if current < (recent_low * 1.05) else "middle"
        })

        # RSI divergence check
        rsi = momentum.RSIIndicator(df["Close"])
        rsi_values = rsi.rsi().values

        if rsi_values[-1] > 70:
            patterns.append({
                "pattern": "RSI Overbought",
                "type": "bearish",
                "rsi": round(rsi_values[-1], 2)
            })
        elif rsi_values[-1] < 30:
            patterns.append({
                "pattern": "RSI Oversold",
                "type": "bullish",
                "rsi": round(rsi_values[-1], 2)
            })

        return json.dumps({
            "symbol": symbol,
            "patterns": patterns
        }, indent=2)

    @tool("Calculate Risk Metrics")
    def calculate_risk_metrics(
        self,
        symbol: str,
        benchmark: str = "SPY",
        period: str = "1y"
    ) -> str:
        """
        Calculate risk metrics for a stock.

        Args:
            symbol: Stock ticker symbol
            benchmark: Benchmark symbol for comparison
            period: Analysis period
        """
        from config.settings import settings

        # Get data
        stock = yf.Ticker(symbol)
        bench = yf.Ticker(benchmark)

        stock_hist = stock.history(period=period)
        bench_hist = bench.history(period=period)

        if stock_hist.empty or bench_hist.empty:
            return json.dumps({"error": "Could not fetch data"})

        # Calculate returns
        stock_returns = stock_hist["Close"].pct_change().dropna()
        bench_returns = bench_hist["Close"].pct_change().dropna()

        # Align dates
        common_dates = stock_returns.index.intersection(bench_returns.index)
        stock_returns = stock_returns[common_dates]
        bench_returns = bench_returns[common_dates]

        metrics = {}

        # Volatility (annualized)
        metrics["volatility"] = round(stock_returns.std() * np.sqrt(252) * 100, 2)

        # Beta
        covariance = np.cov(stock_returns, bench_returns)[0][1]
        bench_variance = bench_returns.var()
        metrics["beta"] = round(covariance / bench_variance, 3)

        # Alpha (annualized)
        stock_annual_return = (1 + stock_returns.mean()) ** 252 - 1
        bench_annual_return = (1 + bench_returns.mean()) ** 252 - 1
        expected_return = settings.risk_free_rate + metrics["beta"] * (bench_annual_return - settings.risk_free_rate)
        metrics["alpha"] = round((stock_annual_return - expected_return) * 100, 2)

        # Sharpe Ratio
        excess_return = stock_annual_return - settings.risk_free_rate
        metrics["sharpe_ratio"] = round(excess_return / (stock_returns.std() * np.sqrt(252)), 3)

        # Sortino Ratio
        downside_returns = stock_returns[stock_returns < 0]
        downside_std = downside_returns.std() * np.sqrt(252)
        metrics["sortino_ratio"] = round(excess_return / downside_std, 3) if downside_std > 0 else None

        # Maximum Drawdown
        cumulative = (1 + stock_returns).cumprod()
        rolling_max = cumulative.expanding().max()
        drawdowns = (cumulative - rolling_max) / rolling_max
        metrics["max_drawdown"] = round(drawdowns.min() * 100, 2)

        # Value at Risk (95%)
        metrics["var_95"] = round(np.percentile(stock_returns, 5) * 100, 2)

        return json.dumps({
            "symbol": symbol,
            "benchmark": benchmark,
            "period": period,
            "metrics": metrics
        }, indent=2)

Building the Agents

Market Data Analyst

# agents/market_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_market_analyst(llm: Ollama, mcp_tools: list) -> Agent:
    return Agent(
        role='Market Data Analyst',
        goal='Gather and interpret market data, price movements, and trading patterns',
        backstory='''You are a quantitative market analyst with deep expertise in
        financial markets. You excel at:
        - Analyzing price and volume data
        - Identifying market trends and patterns
        - Interpreting market indicators
        - Understanding market microstructure
        - Contextualizing movements with market events''',
        llm=llm,
        tools=mcp_tools,
        verbose=True
    )

def create_market_analysis_task(agent: Agent, symbol: str):
    return f"""Analyze the current market data for {symbol}:

1. **Price Analysis**
   - Current price and recent changes
   - Position relative to 52-week range
   - Comparison to key moving averages

2. **Volume Analysis**
   - Current volume vs. average
   - Volume trends
   - Any unusual activity

3. **Market Context**
   - How is the broader market performing?
   - Sector performance
   - Any relevant market events

4. **Key Metrics**
   - Market cap and valuation multiples
   - Volatility measures
   - Liquidity indicators

Provide factual data with brief interpretation."""

Fundamental Analyst

# agents/fundamental_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_fundamental_analyst(llm: Ollama, mcp_tools: list) -> Agent:
    return Agent(
        role='Fundamental Analyst',
        goal='Evaluate company financials, business model, and intrinsic value',
        backstory='''You are a senior fundamental analyst with expertise in
        financial statement analysis and valuation. You:
        - Analyze income statements, balance sheets, and cash flows
        - Calculate and interpret financial ratios
        - Evaluate competitive positioning
        - Assess management quality and strategy
        - Estimate intrinsic value using multiple methods''',
        llm=llm,
        tools=mcp_tools,
        verbose=True
    )

def create_fundamental_analysis_task(agent: Agent, symbol: str):
    return f"""Perform fundamental analysis on {symbol}:

1. **Financial Health**
   - Revenue and earnings trends
   - Profit margins (gross, operating, net)
   - Return on equity and assets
   - Debt levels and coverage ratios

2. **Cash Flow Analysis**
   - Operating cash flow quality
   - Free cash flow generation
   - Capital allocation priorities

3. **Valuation**
   - P/E ratio vs. historical and peers
   - Price-to-book and price-to-sales
   - EV/EBITDA comparison
   - Dividend yield and payout ratio

4. **Business Quality**
   - Competitive advantages
   - Growth drivers
   - Key risks

5. **Fair Value Estimate**
   - Based on your analysis, is the stock:
     - Undervalued
     - Fairly valued
     - Overvalued
   - Provide reasoning"""

Technical Analyst

# agents/technical_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama
from tools.financial_tools import TechnicalAnalysisTools

def create_technical_analyst(llm: Ollama, tools: TechnicalAnalysisTools) -> Agent:
    return Agent(
        role='Technical Analyst',
        goal='Analyze price charts, indicators, and trading patterns',
        backstory='''You are an expert technical analyst who reads charts
        to identify trading opportunities. You:
        - Analyze price patterns and trends
        - Use technical indicators effectively
        - Identify support and resistance levels
        - Time entry and exit points
        - Manage risk with technical tools''',
        llm=llm,
        tools=[
            tools.calculate_indicators,
            tools.identify_patterns,
            tools.calculate_risk_metrics
        ],
        verbose=True
    )

def create_technical_analysis_task(agent: Agent, symbol: str):
    return f"""Perform technical analysis on {symbol}:

1. **Trend Analysis**
   - Primary trend direction
   - Moving average alignment
   - Trend strength

2. **Momentum**
   - RSI reading and interpretation
   - MACD signal
   - Momentum divergences

3. **Volatility**
   - Bollinger Band position
   - ATR for risk sizing
   - Volatility trend

4. **Chart Patterns**
   - Any recognizable patterns
   - Support/resistance levels
   - Breakout potential

5. **Trading Signals**
   - Current technical stance (bullish/bearish/neutral)
   - Key levels to watch
   - Risk/reward assessment"""

Risk Analyst

# agents/risk_analyst.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_risk_analyst(llm: Ollama, tools: list) -> Agent:
    return Agent(
        role='Risk Analyst',
        goal='Assess investment risks and portfolio implications',
        backstory='''You are a risk management specialist focused on
        protecting capital while capturing returns. You:
        - Quantify various risk factors
        - Stress test investment scenarios
        - Consider tail risks and black swans
        - Evaluate portfolio impact
        - Recommend position sizing''',
        llm=llm,
        tools=tools,
        verbose=True
    )

def create_risk_analysis_task(agent: Agent, symbol: str):
    return f"""Analyze investment risks for {symbol}:

1. **Market Risk**
   - Beta and correlation to market
   - Volatility assessment
   - Value at Risk

2. **Company-Specific Risk**
   - Business model risks
   - Financial leverage
   - Management/governance risks

3. **External Risks**
   - Regulatory risks
   - Competitive threats
   - Macroeconomic sensitivity

4. **Downside Scenarios**
   - What could go wrong?
   - Maximum drawdown potential
   - Stress test results

5. **Risk-Adjusted View**
   - Is the potential return worth the risk?
   - Suggested position sizing
   - Hedging considerations"""

Report Compiler

# agents/report_compiler.py
from crewai import Agent
from langchain_community.llms import Ollama

def create_report_compiler(llm: Ollama) -> Agent:
    return Agent(
        role='Investment Report Compiler',
        goal='Synthesize analysis into actionable investment reports',
        backstory='''You are a senior investment strategist who creates
        compelling, actionable research reports. You:
        - Synthesize diverse analytical perspectives
        - Identify the most important factors
        - Communicate clearly for decision-making
        - Provide specific, actionable recommendations
        - Balance conviction with appropriate caveats''',
        llm=llm,
        verbose=True
    )

def create_report_task(agent: Agent, symbol: str):
    return f"""Compile a comprehensive investment report for {symbol}:

## Report Structure

### Executive Summary
- Investment thesis in 2-3 sentences
- Key recommendation (Buy/Hold/Sell)
- Target price if applicable
- Key risks

### Company Overview
- Brief business description
- Competitive position
- Recent developments

### Investment Analysis

#### Fundamental View
(Summarize fundamental analysis)

#### Technical View
(Summarize technical analysis)

#### Risk Assessment
(Summarize key risks)

### Recommendation

**Rating:** [Strong Buy / Buy / Hold / Sell / Strong Sell]

**Conviction Level:** [High / Medium / Low]

**Investment Horizon:** [Short-term / Medium-term / Long-term]

**Key Catalysts:**
- Near-term catalysts
- Long-term value drivers

**Risks to Thesis:**
- Primary risks
- Mitigating factors

### Action Items
- Specific entry/exit recommendations
- Position sizing guidance
- Monitoring points"""

Complete Workflow

Stock Analysis Crew

# workflows/stock_analysis.py
from crewai import Crew, Process, Task
from langchain_community.llms import Ollama
from mcp import ClientSession
from agents.market_analyst import create_market_analyst, create_market_analysis_task
from agents.fundamental_analyst import create_fundamental_analyst, create_fundamental_analysis_task
from agents.technical_analyst import create_technical_analyst, create_technical_analysis_task
from agents.risk_analyst import create_risk_analyst, create_risk_analysis_task
from agents.report_compiler import create_report_compiler, create_report_task
from tools.financial_tools import TechnicalAnalysisTools
from config.settings import settings
import asyncio

class StockAnalyzer:
    def __init__(self):
        # Initialize LLM
        self.llm = Ollama(
            model=settings.ollama_model,
            base_url=settings.ollama_base_url
        )

        # Initialize tools
        self.tech_tools = TechnicalAnalysisTools()

        # MCP tools will be loaded from servers
        self.mcp_tools = []

    async def connect_mcp_servers(self):
        """Connect to MCP servers for financial data."""
        # In production, this would connect to actual MCP servers
        # For now, we use the direct yfinance tools
        pass

    def analyze_stock(self, symbol: str) -> str:
        """Run full stock analysis."""

        # Create agents
        market_analyst = create_market_analyst(self.llm, self.mcp_tools)
        fundamental_analyst = create_fundamental_analyst(self.llm, self.mcp_tools)
        technical_analyst = create_technical_analyst(self.llm, self.tech_tools)
        risk_analyst = create_risk_analyst(self.llm, [self.tech_tools.calculate_risk_metrics])
        report_compiler = create_report_compiler(self.llm)

        # Create tasks
        market_task = Task(
            description=create_market_analysis_task(market_analyst, symbol),
            expected_output="Market data analysis",
            agent=market_analyst
        )

        fundamental_task = Task(
            description=create_fundamental_analysis_task(fundamental_analyst, symbol),
            expected_output="Fundamental analysis",
            agent=fundamental_analyst
        )

        technical_task = Task(
            description=create_technical_analysis_task(technical_analyst, symbol),
            expected_output="Technical analysis",
            agent=technical_analyst
        )

        risk_task = Task(
            description=create_risk_analysis_task(risk_analyst, symbol),
            expected_output="Risk analysis",
            agent=risk_analyst,
            context=[market_task, fundamental_task, technical_task]
        )

        report_task = Task(
            description=create_report_task(report_compiler, symbol),
            expected_output="Investment report",
            agent=report_compiler,
            context=[market_task, fundamental_task, technical_task, risk_task]
        )

        # Create and run crew
        crew = Crew(
            agents=[
                market_analyst,
                fundamental_analyst,
                technical_analyst,
                risk_analyst,
                report_compiler
            ],
            tasks=[
                market_task,
                fundamental_task,
                technical_task,
                risk_task,
                report_task
            ],
            process=Process.sequential,
            verbose=True
        )

        result = crew.kickoff()
        return str(result)

    def quick_analysis(self, symbol: str) -> str:
        """Quick analysis focusing on key metrics."""

        # Simplified analysis with fewer agents
        technical_analyst = create_technical_analyst(self.llm, self.tech_tools)
        report_compiler = create_report_compiler(self.llm)

        quick_analysis_task = Task(
            description=f"""Quick analysis of {symbol}:
            1. Calculate key technical indicators
            2. Identify current trend
            3. Note key support/resistance levels
            4. Provide a brief trading view""",
            expected_output="Quick technical overview",
            agent=technical_analyst
        )

        brief_report_task = Task(
            description=f"""Create a brief report for {symbol}:
            - Current technical stance
            - Key levels
            - Near-term outlook
            Keep it under 300 words.""",
            expected_output="Brief report",
            agent=report_compiler,
            context=[quick_analysis_task]
        )

        crew = Crew(
            agents=[technical_analyst, report_compiler],
            tasks=[quick_analysis_task, brief_report_task],
            process=Process.sequential,
            verbose=True
        )

        return str(crew.kickoff())

    def compare_stocks(self, symbols: list[str]) -> str:
        """Compare multiple stocks."""

        market_analyst = create_market_analyst(self.llm, self.mcp_tools)
        report_compiler = create_report_compiler(self.llm)

        comparison_task = Task(
            description=f"""Compare these stocks: {', '.join(symbols)}

            For each stock, gather:
            1. Current price and valuation
            2. Key financial metrics
            3. Technical position

            Then compare:
            - Which is cheapest by valuation?
            - Which has strongest momentum?
            - Which has best risk/reward?""",
            expected_output="Stock comparison data",
            agent=market_analyst
        )

        ranking_task = Task(
            description=f"""Based on the comparison, rank {', '.join(symbols)}:
            1. Best overall investment
            2. Best value play
            3. Best momentum play
            4. Most risky

            Explain your rankings.""",
            expected_output="Ranked comparison report",
            agent=report_compiler,
            context=[comparison_task]
        )

        crew = Crew(
            agents=[market_analyst, report_compiler],
            tasks=[comparison_task, ranking_task],
            process=Process.sequential,
            verbose=True
        )

        return str(crew.kickoff())

Main Entry Point

# main.py
from workflows.stock_analysis import StockAnalyzer
from datetime import datetime
import argparse

def main():
    parser = argparse.ArgumentParser(description="Financial Analyst Agent")
    parser.add_argument("symbol", help="Stock symbol to analyze")
    parser.add_argument(
        "--mode",
        choices=["full", "quick", "compare"],
        default="full",
        help="Analysis mode"
    )
    parser.add_argument(
        "--compare-with",
        nargs="+",
        help="Additional symbols for comparison mode"
    )
    parser.add_argument("--output", help="Output file path")

    args = parser.parse_args()

    analyzer = StockAnalyzer()

    print(f"Analyzing {args.symbol}...")
    print("-" * 50)

    if args.mode == "full":
        result = analyzer.analyze_stock(args.symbol)
    elif args.mode == "quick":
        result = analyzer.quick_analysis(args.symbol)
    elif args.mode == "compare":
        symbols = [args.symbol] + (args.compare_with or [])
        result = analyzer.compare_stocks(symbols)

    # Save output
    if args.output:
        output_path = args.output
    else:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_path = f"output/{args.symbol}_{args.mode}_{timestamp}.md"

    with open(output_path, "w") as f:
        f.write(result)

    print(f"\nAnalysis complete! Report saved to: {output_path}")

if __name__ == "__main__":
    main()

Usage Examples

# Full analysis
python main.py AAPL --mode full

# Quick analysis
python main.py MSFT --mode quick

# Compare stocks
python main.py AAPL --mode compare --compare-with MSFT GOOGL AMZN

# Specify output
python main.py TSLA --output reports/tesla_analysis.md

Summary

Gathers comprehensive data: Market prices, financials, technical indicators
Analyzes from multiple angles: Fundamental, technical, and risk perspectives
Synthesizes actionable insights: Clear recommendations with supporting analysis
Runs locally: Using Ollama for cost-effective inference

Key implementation points:

Specialize agents by analysis type: Each perspective gets dedicated focus
Use MCP for data access: Structured, reliable financial data retrieval
Layer analysis logically: Market data, then fundamental, then technical, then risk
Produce actionable output: Reports should support decision-making

This architecture extends naturally to portfolio analysis, market screening, and automated monitoring. The multi-agent foundation makes these extensions straightforward to implement.

This concludes the agent tutorials. Ready to learn about AI skills? Continue to What is an AI Skill? to understand how skills complement agents in the AI development toolkit.