# Directory Structure
```
├── .gitignore
├── assets
│ └── logo.png
├── claude_server.py
├── main.py
├── portfolio_server
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── alpha_vantage.py
│ │ └── news_api.py
│ ├── data
│ │ ├── __init__.py
│ │ ├── portfolio.py
│ │ └── storage.py
│ ├── resources
│ │ ├── __init__.py
│ │ └── portfolio_resources.py
│ ├── server.py
│ ├── sse.py
│ └── tools
│ ├── __init__.py
│ ├── analysis_tools.py
│ ├── portfolio_tools.py
│ ├── stock_tools.py
│ └── visualization_tools.py
├── README.md
└── requirements.txt
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | .env
2 | venv
3 | .venv
4 | __pycache__
5 | *.pyc
6 | *.pyo
7 | *.pyd
8 | *.db
9 | .DS_Store
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Portfolio Manager MCP Server
2 | <p align="center">
3 | <img src="assets/logo.png" width="300" alt="Project Logo">
4 | </p>
5 |
6 | <p align="center">
7 | <a href="https://mseep.ai/app/ikhyunan-mcp-investmentportfolio">
8 | <img src="https://mseep.net/pr/ikhyunan-mcp-investmentportfolio-badge.png" alt="MseeP.ai Security Assessment Badge" />
9 | </a>
10 | </p>
11 |
12 |
13 | A Model Context Protocol (MCP) server that provides tools and resources for managing and analyzing investment portfolios.
14 |
15 | ## Features
16 |
17 | - **Portfolio Management**: Create and update investment portfolios with stocks and bonds
18 | - **Market Data**: Fetch real-time stock price information and relevant news
19 | - **Analysis**: Generate comprehensive portfolio reports and performance analysis
20 | - **Recommendations**: Get personalized investment recommendations based on portfolio composition
21 | - **Visualization**: Create visual representations of portfolio allocation
22 |
23 | ## Installation
24 |
25 | 1. Clone this repository:
26 | ```bash
27 | git clone https://github.com/ikhyunAn/portfolio-manager-mcp.git
28 | cd portfolio-manager-mcp
29 | ```
30 |
31 | 2. Install the required dependencies:
32 | ```bash
33 | pip install -r requirements.txt
34 | ```
35 |
36 | 3. Set up API keys (optional):
37 | ```bash
38 | export ALPHA_VANTAGE_API_KEY="your_key_here"
39 | export NEWS_API_KEY="your_key_here"
40 | ```
41 |
42 | Alternatively, create a `.env` file in the root of the directory and store the API keys
43 |
44 | ## Usage
45 |
46 | ### Running the Server
47 |
48 | You can run the server in two different modes:
49 |
50 | 1. **Stdio Transport** (default, for Claude Desktop integration):
51 | ```bash
52 | python main.py # alternate commands: i.e.) python3, python3.11
53 | ```
54 |
55 | 2. **SSE Transport** (for HTTP-based clients):
56 | ```bash
57 | python main.py --sse
58 | ```
59 |
60 | ### Integration with Claude Desktop
61 |
62 | Add the server to your Claude Desktop configuration file:
63 |
64 | ```json
65 | {
66 | "mcpServers": {
67 | "portfolio-manager": {
68 | "command": "python", // may use different command
69 | "args": ["/path/to/portfolio-manager-mcp/main.py"],
70 | "env": {
71 | "ALPHA_VANTAGE_API_KEY": "your_key_here",
72 | "NEWS_API_KEY": "your_key_here"
73 | }
74 | }
75 | }
76 | }
77 | ```
78 |
79 | If you choose to run your server in a virtual environment, then your configuration file will look like:
80 |
81 | ```json
82 | {
83 | "mcpServers": {
84 | "portfolio-manager": {
85 | "command": "/path/to/portfolio-manager-mcp/venv/bin/python",
86 | "args": ["/path/to/portfolio-manager-mcp/main.py"],
87 | "env": {
88 | "PYTHONPATH": "/path/to/portfolio-manager-mcp",
89 | "ALPHA_VANTAGE_API_KEY": "your_key_here",
90 | "NEWS_API_KEY": "your_key_here"
91 | }
92 | }
93 | }
94 | }
95 | ```
96 |
97 | To run it in a virtual environment:
98 |
99 | ```bash
100 | # Create a virtual environment
101 | python3 -m venv venv
102 |
103 | # Activate the virtual environment
104 | source venv/bin/activate # On macOS/Linux
105 | # or
106 | # venv\Scripts\activate # On Windows
107 |
108 | # Install dependencies
109 | pip install -r requirements.txt
110 |
111 | # Run the server
112 | python3 main.py
113 | ```
114 |
115 |
116 | Or use the MCP CLI for easier installation:
117 |
118 | ```bash
119 | mcp install main.py
120 | ```
121 |
122 | ## Example Queries
123 |
124 | Once the server is running and connected to Claude, you can interact with it using natural language:
125 |
126 | - "Create a portfolio with 30% AAPL, 20% MSFT, 15% AMZN, and 35% US Treasury bonds with user Id <User_ID>"
127 | - "What's the recent performance of my portfolio?"
128 | - "Show me news about the stocks in my portfolio"
129 | - "Generate investment recommendations for my current portfolio"
130 | - "Visualize my current asset allocation"
131 |
132 | ## Project Structure
133 |
134 | ```
135 | portfolio-manager/
136 | ├── main.py # Entry point
137 | ├── portfolio_server/ # Main package
138 | │ ├── api/ # External API clients
139 | │ │ ├── alpha_vantage.py # Stock market data API
140 | │ │ └── news_api.py # News API
141 | │ ├── data/ # Data management
142 | │ │ ├── portfolio.py # Portfolio models
143 | │ │ └── storage.py # Data persistence
144 | │ ├── resources/ # MCP resources
145 | │ │ └── portfolio_resources.py # Portfolio resource definitions
146 | │ ├── tools/ # MCP tools
147 | │ │ ├── analysis_tools.py # Portfolio analysis
148 | │ │ ├── portfolio_tools.py # Portfolio management
149 | │ │ ├── stock_tools.py # Stock data and news
150 | │ │ └── visualization_tools.py # Visualization tools
151 | │ └── server.py # MCP server setup
152 | └── requirements.txt # Dependencies
153 | ```
154 |
155 | ## Future Work
156 |
157 | As of now, the MCP program uses manually created JSON file which keeps track of each user's investment portfolio.
158 |
159 | This should be fixed so that it reads in the portfolio data from actual banking applications.
160 |
161 |
162 | ### Tasks
163 |
164 | - [ ] Extract JSON from a Finance or Banking Application which the user uses
165 | - [ ] Enable modifying the investment portfolio by the client
166 | - [ ] Implement automated portfolio rebalancing
167 | - [ ] Add support for cryptocurrency assets
168 | - [ ] Develop mobile application integration
169 |
170 | ## License
171 |
172 | MIT
173 |
```
--------------------------------------------------------------------------------
/portfolio_server/resources/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """MCP resources for portfolio data."""
2 |
3 | from portfolio_server.resources import portfolio_resources
4 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | mcp[cli]>=1.5.0
2 | pandas>=2.0.0
3 | httpx>=0.25.0
4 | matplotlib>=3.7.0
5 | uvicorn>=0.25.0
6 | starlette>=0.36.0
7 | asyncio>=3.4.3
8 |
```
--------------------------------------------------------------------------------
/portfolio_server/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Portfolio Manager MCP Server
3 |
4 | A Model Context Protocol (MCP) server for managing investment portfolios.
5 | """
6 |
7 | __version__ = "0.1.0"
8 |
```
--------------------------------------------------------------------------------
/portfolio_server/data/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Data management utilities for portfolios."""
2 |
3 | from portfolio_server.data.storage import (
4 | get_portfolio_path,
5 | load_portfolio,
6 | save_portfolio,
7 | )
8 |
```
--------------------------------------------------------------------------------
/portfolio_server/api/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """API clients for external services."""
2 |
3 | from portfolio_server.api.alpha_vantage import fetch_stock_data
4 | from portfolio_server.api.news_api import fetch_stock_news
5 |
```
--------------------------------------------------------------------------------
/portfolio_server/tools/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """MCP tools for portfolio management and analysis."""
2 |
3 | # Import all tools to make them available when importing this package
4 | from portfolio_server.tools import (
5 | portfolio_tools,
6 | stock_tools,
7 | analysis_tools,
8 | visualization_tools,
9 | )
10 |
```
--------------------------------------------------------------------------------
/portfolio_server/api/news_api.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | News API client for fetching stock news.
3 | """
4 | import os
5 | import httpx
6 | from typing import Dict, Any, List
7 |
8 | NEWS_API_KEY = os.environ.get("NEWS_API_KEY", "demo")
9 |
10 | async def fetch_stock_news(symbol: str, max_articles: int = 5) -> List[Dict[str, Any]]:
11 | """
12 | Fetch news articles about a specific stock
13 |
14 | Args:
15 | symbol: Stock symbol to get news for
16 | max_articles: Maximum number of articles to return
17 |
18 | Returns:
19 | List of news article data
20 | """
21 | url = f"https://newsapi.org/v2/everything?q={symbol}&apiKey={NEWS_API_KEY}&sortBy=publishedAt&language=en&pageSize={max_articles}"
22 |
23 | async with httpx.AsyncClient() as client:
24 | response = await client.get(url)
25 | data = response.json()
26 |
27 | if data.get("status") != "ok":
28 | return [{"error": data.get("message", "Unknown error")}]
29 |
30 | articles = []
31 | for article in data.get("articles", [])[:max_articles]:
32 | articles.append({
33 | "title": article.get("title"),
34 | "source": article.get("source", {}).get("name"),
35 | "url": article.get("url"),
36 | "published_at": article.get("publishedAt"),
37 | "description": article.get("description")
38 | })
39 |
40 | return articles
41 |
```
--------------------------------------------------------------------------------
/portfolio_server/data/storage.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Storage utilities for portfolio data.
3 | """
4 | import os
5 | import json
6 | from datetime import datetime
7 | from typing import Dict, Any
8 |
9 | # Setup storage paths
10 | PORTFOLIO_DIR = os.path.expanduser("~/.portfolio-manager")
11 | os.makedirs(PORTFOLIO_DIR, exist_ok=True)
12 |
13 | def get_portfolio_path(user_id: str) -> str:
14 | """
15 | Get the path to a user's portfolio file.
16 |
17 | Args:
18 | user_id: Unique identifier for the user
19 |
20 | Returns:
21 | Path to the portfolio JSON file
22 | """
23 | return os.path.join(PORTFOLIO_DIR, f"{user_id}_portfolio.json")
24 |
25 | def load_portfolio(user_id: str) -> Dict[str, Any]:
26 | """
27 | Load a user's portfolio, or return empty one if none exists.
28 |
29 | Args:
30 | user_id: Unique identifier for the user
31 |
32 | Returns:
33 | Portfolio data including stocks and bonds
34 | """
35 | path = get_portfolio_path(user_id)
36 | if os.path.exists(path):
37 | with open(path, 'r') as f:
38 | return json.load(f)
39 | return {"stocks": {}, "bonds": {}, "last_updated": None}
40 |
41 | def save_portfolio(user_id: str, portfolio: Dict[str, Any]) -> None:
42 | """
43 | Save a user's portfolio to disk.
44 |
45 | Args:
46 | user_id: Unique identifier for the user
47 | portfolio: Portfolio data to save
48 | """
49 | portfolio["last_updated"] = datetime.now().isoformat()
50 | with open(get_portfolio_path(user_id), 'w') as f:
51 | json.dump(portfolio, f, indent=2)
52 |
```
--------------------------------------------------------------------------------
/portfolio_server/data/portfolio.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Portfolio data models and business logic.
3 | """
4 | from typing import Dict, List, Optional, Union
5 |
6 | class Portfolio:
7 | """
8 | Represents a user's investment portfolio.
9 | """
10 | def __init__(self, stocks: Dict[str, float] = None, bonds: Dict[str, float] = None):
11 | self.stocks = stocks or {}
12 | self.bonds = bonds or {}
13 |
14 | @property
15 | def stock_allocation(self) -> float:
16 | """Get the total percentage allocated to stocks."""
17 | return sum(self.stocks.values())
18 |
19 | @property
20 | def bond_allocation(self) -> float:
21 | """Get the total percentage allocated to bonds."""
22 | return sum(self.bonds.values())
23 |
24 | @property
25 | def total_allocation(self) -> float:
26 | """Get the total percentage allocated."""
27 | return self.stock_allocation + self.bond_allocation
28 |
29 | def is_valid(self) -> bool:
30 | """Check if the portfolio allocations sum to approximately 100%."""
31 | return 95 <= self.total_allocation <= 105
32 |
33 | def to_dict(self) -> Dict:
34 | """Convert portfolio to a dictionary representation."""
35 | return {
36 | "stocks": self.stocks,
37 | "bonds": self.bonds,
38 | }
39 |
40 | @classmethod
41 | def from_dict(cls, data: Dict) -> 'Portfolio':
42 | """Create a Portfolio instance from a dictionary."""
43 | return cls(
44 | stocks=data.get("stocks", {}),
45 | bonds=data.get("bonds", {})
46 | )
47 |
```
--------------------------------------------------------------------------------
/portfolio_server/resources/portfolio_resources.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | MCP resources for portfolio data.
3 | """
4 | import json
5 | from typing import List
6 |
7 | from portfolio_server.data.storage import load_portfolio
8 | from portfolio_server.tools.stock_tools import get_stock_prices
9 |
10 | def get_portfolio_resource(user_id: str) -> str:
11 | """
12 | Get the current portfolio data as a resource
13 |
14 | Args:
15 | user_id: Unique identifier for the user
16 | """
17 | portfolio = load_portfolio(user_id)
18 | return json.dumps(portfolio, indent=2)
19 |
20 | async def get_portfolio_performance(user_id: str) -> str:
21 | """
22 | Get the current portfolio performance data as a resource
23 |
24 | Args:
25 | user_id: Unique identifier for the user
26 | """
27 | portfolio = load_portfolio(user_id)
28 |
29 | if not portfolio["stocks"]:
30 | return "No stocks in portfolio to analyze performance."
31 |
32 | # Get stock price data
33 | stock_symbols = list(portfolio["stocks"].keys())
34 | price_data = json.loads(await get_stock_prices(stock_symbols))
35 |
36 | # Calculate performance metrics
37 | performance = {
38 | "symbols": {},
39 | "total_contribution": 0
40 | }
41 |
42 | for symbol, allocation in portfolio["stocks"].items():
43 | if symbol in price_data and "percent_change" in price_data[symbol]:
44 | change = price_data[symbol]["percent_change"]
45 | contribution = (change * allocation) / 100
46 | performance["symbols"][symbol] = {
47 | "allocation": allocation,
48 | "percent_change": change,
49 | "contribution": contribution
50 | }
51 | performance["total_contribution"] += contribution
52 |
53 | return json.dumps(performance, indent=2)
54 |
```
--------------------------------------------------------------------------------
/portfolio_server/tools/visualization_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tools for visualizing portfolio data.
3 | """
4 | import matplotlib.pyplot as plt
5 | from io import BytesIO
6 | from mcp.server.fastmcp import Image
7 |
8 | from portfolio_server.data.storage import load_portfolio
9 |
10 | def visualize_portfolio(user_id: str) -> Image:
11 | """
12 | Create a visualization of the current portfolio allocation
13 |
14 | Args:
15 | user_id: Unique identifier for the user
16 | """
17 | portfolio = load_portfolio(user_id)
18 |
19 | # Prepare data for visualization
20 | labels = []
21 | sizes = []
22 | colors = []
23 |
24 | # Add stocks (in blue shades)
25 | for i, (symbol, allocation) in enumerate(portfolio["stocks"].items()):
26 | labels.append(f"{symbol} ({allocation}%)")
27 | sizes.append(allocation)
28 | # Generate different blue shades
29 | blue_val = min(0.8, 0.3 + (i * 0.1))
30 | colors.append((0, 0, blue_val))
31 |
32 | # Add bonds (in green shades)
33 | for i, (bond_id, allocation) in enumerate(portfolio["bonds"].items()):
34 | labels.append(f"{bond_id} ({allocation}%)")
35 | sizes.append(allocation)
36 | # Generate different green shades
37 | green_val = min(0.8, 0.3 + (i * 0.1))
38 | colors.append((0, green_val, 0))
39 |
40 | # Create pie chart
41 | plt.figure(figsize=(10, 7))
42 | plt.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%', startangle=140)
43 | plt.axis('equal') # Equal aspect ratio ensures the pie chart is circular
44 | plt.title(f"Portfolio Allocation for User {user_id}")
45 |
46 | # Save to bytes object
47 | buf = BytesIO()
48 | plt.savefig(buf, format='png')
49 | buf.seek(0)
50 | plt.close()
51 |
52 | # Return as MCP Image
53 | return Image(data=buf.getvalue(), format="png")
54 |
```
--------------------------------------------------------------------------------
/portfolio_server/api/alpha_vantage.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Alpha Vantage API client for fetching stock data.
3 | """
4 | import os
5 | import httpx
6 | from typing import Dict, Any, List
7 |
8 | ALPHA_VANTAGE_API_KEY = os.environ.get("ALPHA_VANTAGE_API_KEY", "demo")
9 |
10 | async def fetch_stock_data(symbol: str) -> Dict[str, Any]:
11 | """
12 | Fetch stock data from Alpha Vantage API
13 |
14 | Args:
15 | symbol: Stock symbol to fetch data for
16 |
17 | Returns:
18 | Dictionary with stock price data
19 | """
20 | url = f"https://www.alphavantage.co/query?function=TIME_SERIES_DAILY&symbol={symbol}&apikey={ALPHA_VANTAGE_API_KEY}"
21 |
22 | async with httpx.AsyncClient() as client:
23 | response = await client.get(url)
24 | data = response.json()
25 | return data
26 |
27 | async def search_company(query: str) -> List[Dict[str, str]]:
28 | """
29 | Search for companies by name or symbol using Alpha Vantage API
30 |
31 | Args:
32 | query: Company name or symbol to search for
33 |
34 | Returns:
35 | List of dictionaries containing company information:
36 | {
37 | "symbol": "AAPL",
38 | "name": "Apple Inc",
39 | "type": "Equity",
40 | "region": "United States"
41 | }
42 | """
43 | url = f"https://www.alphavantage.co/query?function=SYMBOL_SEARCH&keywords={query}&apikey={ALPHA_VANTAGE_API_KEY}"
44 |
45 | async with httpx.AsyncClient() as client:
46 | response = await client.get(url)
47 | data = response.json()
48 |
49 | if "bestMatches" not in data:
50 | return []
51 |
52 | results = []
53 | for match in data["bestMatches"]:
54 | results.append({
55 | "symbol": match["1. symbol"],
56 | "name": match["2. name"],
57 | "type": match["3. type"],
58 | "region": match["4. region"]
59 | })
60 |
61 | return results
62 |
```
--------------------------------------------------------------------------------
/portfolio_server/tools/portfolio_tools.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Dict, List, Optional
2 | from mcp.server.fastmcp import Context
3 | from portfolio_server.data.storage import load_portfolio, save_portfolio
4 |
5 | def update_portfolio(user_id: str,
6 | stocks: Optional[Dict[str, float]] = None,
7 | bonds: Optional[Dict[str, float]] = None,
8 | ctx: Context = None) -> str:
9 |
10 | portfolio = load_portfolio(user_id)
11 |
12 | if stocks:
13 | portfolio["stocks"].update(stocks)
14 | if bonds:
15 | portfolio["bonds"].update(bonds)
16 |
17 | # validate accuracy of the portfolio
18 | total_percent = sum(portfolio["stocks"].values()) + sum(portfolio["bonds"].values())
19 | if not (95 <= total_percent <= 105):
20 | return f"Warning: Total allocation is {total_percent}%, which is not close to 100%"
21 |
22 | save_portfolio(user_id, portfolio)
23 |
24 | # return the updated portfolio
25 | return f"Portfolio updated successfully for user {user_id}." \
26 | f"({len(portfolio['stocks'])} stocks, {len(portfolio['bonds'])} bonds)"
27 |
28 | def remove_investment(user_id: str,
29 | stock_symbols: Optional[List[str]] = None,
30 | bond_ids: Optional[List[str]] = None) -> str:
31 |
32 | portfolio = load_portfolio(user_id)
33 |
34 | removed = []
35 | if stock_symbols:
36 | for symbol in stock_symbols:
37 | if symbol in portfolio["stocks"]:
38 | del portfolio["stocks"][symbol]
39 | removed.append(symbol)
40 |
41 | if bond_ids:
42 | for bond_id in bond_ids:
43 | if bond_id in portfolio["bonds"]:
44 | del portfolio["bonds"][bond_id]
45 | removed.append(bond_id)
46 |
47 | save_portfolio(user_id, portfolio)
48 |
49 | if removed:
50 | return f"Removed investments: {', '.join(removed)} from user {user_id}'s portfolio."
51 | else:
52 | return "No matching investments are found for removal."
53 |
54 |
```
--------------------------------------------------------------------------------
/portfolio_server/server.py:
--------------------------------------------------------------------------------
```python
1 | import sys
2 |
3 | from mcp.server.fastmcp import FastMCP
4 | from portfolio_server.tools import portfolio_tools, stock_tools, analysis_tools, visualization_tools
5 | from portfolio_server.resources import portfolio_resources
6 |
7 | def create_mcp_server() -> FastMCP:
8 | # Create and configure the MCP server with default transport (stdio)
9 | try:
10 | print("Initializing Portfolio Manager MCP Server...", file=sys.stderr)
11 | mcp = FastMCP("Portfolio Manager MCP Server",
12 | dependencies = [
13 | "pandas",
14 | "httpx",
15 | "matplotlib"
16 | ])
17 |
18 | # Register tools
19 | print("Registering tools...", file=sys.stderr)
20 | register_tools(mcp)
21 | print("Registering resources...", file=sys.stderr)
22 | register_resources(mcp)
23 |
24 | print("MCP Server initialized successfully!", file=sys.stderr)
25 | return mcp
26 | except Exception as e:
27 | print(f"ERROR initializing MCP server: {str(e)}", file=sys.stderr)
28 | print(f"Error type: {type(e)}", file=sys.stderr)
29 | # Re-raise the exception so it's visible in the logs
30 | raise
31 |
32 | def register_tools(mcp: FastMCP) -> None:
33 | # TODO: Register all tools
34 | mcp.tool()(portfolio_tools.update_portfolio)
35 | mcp.tool()(portfolio_tools.remove_investment)
36 |
37 | mcp.tool()(stock_tools.get_stock_prices)
38 | mcp.tool()(stock_tools.get_stock_news)
39 | mcp.tool()(stock_tools.search_stocks)
40 |
41 | mcp.tool()(analysis_tools.generate_portfolio_report)
42 | mcp.tool()(analysis_tools.get_investment_recommendations)
43 |
44 | mcp.tool()(visualization_tools.visualize_portfolio)
45 |
46 | def register_resources(mcp: FastMCP) -> None:
47 | # Register all resources within the portfolio_resources module
48 | mcp.resource("portfolio://{user_id}")(portfolio_resources.get_portfolio_resource)
49 | mcp.resource("portfolio-performance://{user_id}")(portfolio_resources.get_portfolio_performance)
```
--------------------------------------------------------------------------------
/portfolio_server/sse.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | SSE server setup for MCP.
3 | This module provides utilities for running the MCP server with SSE transport.
4 | """
5 | import uvicorn
6 | from starlette.applications import Starlette
7 | from starlette.routing import Mount, Route
8 | from starlette.middleware import Middleware
9 | from starlette.middleware.cors import CORSMiddleware
10 |
11 | from mcp.server.sse import SseServerTransport
12 | from portfolio_server.server import create_mcp_server
13 |
14 | def create_sse_app(port=8080):
15 | """
16 | Create a Starlette app for SSE transport with the portfolio MCP server.
17 |
18 | Args:
19 | port: Port to use for the SSE server
20 |
21 | Returns:
22 | Starlette app configured with SSE routes
23 | """
24 | # Create the MCP server
25 | mcp = create_mcp_server()
26 |
27 | # Create SSE transport
28 | transport = SseServerTransport("/mcp/messages")
29 |
30 | # Define route handlers
31 | async def handle_sse(request):
32 | """Handle SSE connection requests"""
33 | async with transport.connect_sse(
34 | request.scope, request.receive, request._send
35 | ) as streams:
36 | await mcp._mcp_server.run(
37 | streams[0],
38 | streams[1],
39 | mcp._mcp_server.create_initialization_options()
40 | )
41 |
42 | # Create routes
43 | routes = [
44 | Route("/mcp/sse", endpoint=handle_sse),
45 | Mount("/mcp/messages", app=transport.handle_post_message)
46 | ]
47 |
48 | # Configure middleware
49 | middleware = [
50 | Middleware(
51 | CORSMiddleware,
52 | allow_origins=["*"],
53 | allow_methods=["*"],
54 | allow_headers=["*"]
55 | )
56 | ]
57 |
58 | # Create the Starlette app
59 | return Starlette(routes=routes, middleware=middleware)
60 |
61 | def run_sse_server(port=8080, host="0.0.0.0"):
62 | """
63 | Run the MCP server with SSE transport using uvicorn.
64 |
65 | Args:
66 | port: Port to run the server on
67 | host: Host to bind to
68 | """
69 | app = create_sse_app(port)
70 | uvicorn.run(app, host=host, port=port)
71 |
72 | if __name__ == "__main__":
73 | run_sse_server()
74 |
```
--------------------------------------------------------------------------------
/portfolio_server/tools/stock_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tools for retrieving stock data and news.
3 | """
4 | import json
5 | from typing import List, Dict, Any
6 |
7 | from portfolio_server.api.alpha_vantage import fetch_stock_data, search_company
8 | from portfolio_server.api.news_api import fetch_stock_news as fetch_news
9 |
10 | async def _fetch_stock_data_with_fallback(symbol: str, days: int) -> Dict[str, Any]:
11 | """
12 | Helper function to fetch stock data with company name fallback
13 |
14 | Args:
15 | symbol: Stock symbol or company name to fetch data for
16 | days: Number of days of history to include
17 | """
18 | # First try direct symbol lookup
19 | data = await fetch_stock_data(symbol)
20 |
21 | # If direct lookup fails, try searching by company name
22 | if "Error Message" in data or "Time Series (Daily)" not in data:
23 | search_results = await search_company(symbol)
24 |
25 | if search_results:
26 | # Use the first match's symbol
27 | best_match = search_results[0]["symbol"]
28 | data = await fetch_stock_data(best_match)
29 |
30 | # If data is still not available after searching
31 | if "Error Message" in data or "Time Series (Daily)" not in data:
32 | return {
33 | "error": f"Stock Data Not Found. Original Query: {symbol}, Tried Symbol: {best_match}"
34 | }
35 | else:
36 | return {
37 | "error": f"No Company or stock symbol matching '{symbol}' was found."
38 | }
39 |
40 | # Process successful data
41 | if "Time Series (Daily)" in data:
42 | time_series = data["Time Series (Daily)"]
43 | dates = sorted(time_series.keys(), reverse=True)[:days]
44 |
45 | prices = {}
46 | for date in dates:
47 | prices[date] = {
48 | "open": float(time_series[date]["1. open"]),
49 | "high": float(time_series[date]["2. high"]),
50 | "low": float(time_series[date]["3. low"]),
51 | "close": float(time_series[date]["4. close"]),
52 | "volume": int(time_series[date]["5. volume"])
53 | }
54 |
55 | # Calculate change from first to last day
56 | if len(dates) >= 2:
57 | first_close = prices[dates[-1]]["close"]
58 | last_close = prices[dates[0]]["close"]
59 | percent_change = ((last_close - first_close) / first_close) * 100
60 | else:
61 | percent_change = 0
62 |
63 | return {
64 | "prices": prices,
65 | "percent_change": round(percent_change, 2)
66 | }
67 |
68 | return {"error": "Unable to obtain Stock Data"}
69 |
70 | async def get_stock_prices(symbols: List[str], days: int = 7) -> str:
71 | """
72 | Get recent price data for multiple stocks
73 |
74 | Args:
75 | symbols: List of stock symbols or company names to fetch data for
76 | days: Number of days of history to include (default: 7)
77 | """
78 | result = {}
79 |
80 | for symbol in symbols:
81 | result[symbol] = await _fetch_stock_data_with_fallback(symbol, days)
82 |
83 | return json.dumps(result, indent=2)
84 |
85 | async def get_stock_news(symbols: List[str], max_articles: int = 5) -> str:
86 | """
87 | Get recent news articles about stocks in the portfolio
88 |
89 | Args:
90 | symbols: List of stock symbols to get news for
91 | max_articles: Maximum number of articles to return per symbol
92 | """
93 | result = {}
94 |
95 | for symbol in symbols:
96 | articles = await fetch_news(symbol, max_articles)
97 | result[symbol] = articles
98 |
99 | return json.dumps(result, indent=2)
100 |
101 | async def search_stocks(query: str) -> str:
102 | """
103 | Search for stocks by company name or symbol
104 |
105 | Args:
106 | query: Company name or symbol to search for
107 |
108 | Returns:
109 | JSON string containing search results with company information
110 | """
111 | results = await search_company(query)
112 | return json.dumps({"results": results}, indent=2)
113 |
```
--------------------------------------------------------------------------------
/claude_server.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Simplified portfolio manager MCP server for Claude Desktop.
4 | Designed with minimal dependencies to maximize compatibility.
5 | """
6 | import sys
7 | import os
8 | import json
9 | from datetime import datetime
10 | from typing import Dict, List, Optional
11 |
12 | try:
13 | import mcp
14 | from mcp.server.fastmcp import FastMCP
15 | except ImportError:
16 | print("ERROR: MCP package not found. Please install it with: pip install mcp[cli]", file=sys.stderr)
17 | sys.exit(1)
18 |
19 | print(f"Python version: {sys.version}", file=sys.stderr)
20 | print(f"MCP version: {mcp.__version__ if hasattr(mcp, '__version__') else 'unknown'}", file=sys.stderr)
21 | print(f"Current directory: {os.getcwd()}", file=sys.stderr)
22 |
23 | # Create the MCP server
24 | mcp_server = FastMCP("Portfolio Manager")
25 |
26 | # Setup basic storage
27 | PORTFOLIO_DIR = os.path.expanduser("~/.portfolio-manager")
28 | os.makedirs(PORTFOLIO_DIR, exist_ok=True)
29 |
30 | def get_portfolio_path(user_id: str) -> str:
31 | return os.path.join(PORTFOLIO_DIR, f"{user_id}_portfolio.json")
32 |
33 | def load_portfolio(user_id: str) -> Dict:
34 | path = get_portfolio_path(user_id)
35 | if os.path.exists(path):
36 | try:
37 | with open(path, 'r') as f:
38 | return json.load(f)
39 | except json.JSONDecodeError:
40 | print(f"Warning: Invalid JSON in {path}", file=sys.stderr)
41 | return {"stocks": {}, "bonds": {}, "last_updated": None}
42 | return {"stocks": {}, "bonds": {}, "last_updated": None}
43 |
44 | def save_portfolio(user_id: str, portfolio: Dict) -> None:
45 | portfolio["last_updated"] = datetime.now().isoformat()
46 | path = get_portfolio_path(user_id)
47 | with open(path, 'w') as f:
48 | json.dump(portfolio, f, indent=2)
49 |
50 | # Define basic tools
51 | @mcp_server.tool()
52 | def update_portfolio(user_id: str,
53 | stocks: Optional[Dict[str, float]] = None,
54 | bonds: Optional[Dict[str, float]] = None) -> str:
55 | """
56 | Update a user's investment portfolio
57 |
58 | Args:
59 | user_id: Unique identifier for the user
60 | stocks: Dictionary of stock symbols to allocation percentage (e.g. {"AAPL": 10.5, "MSFT": 15.0})
61 | bonds: Dictionary of bond identifiers to allocation percentage (e.g. {"US10Y": 30.0, "CORP_AAA": 20.0})
62 | """
63 | portfolio = load_portfolio(user_id)
64 |
65 | if stocks:
66 | portfolio["stocks"].update(stocks)
67 |
68 | if bonds:
69 | portfolio["bonds"].update(bonds)
70 |
71 | # Validate that percentages sum to approximately 100%
72 | total_pct = sum(portfolio["stocks"].values()) + sum(portfolio["bonds"].values())
73 | if not (95 <= total_pct <= 105):
74 | return f"Warning: Total allocation is {total_pct}%, which is not close to 100%"
75 |
76 | save_portfolio(user_id, portfolio)
77 |
78 | # Return a summary of the updated portfolio
79 | return f"Portfolio updated for user {user_id}. Current allocation: {total_pct}% allocated " \
80 | f"({len(portfolio['stocks'])} stocks, {len(portfolio['bonds'])} bonds)"
81 |
82 | @mcp_server.tool()
83 | def view_portfolio(user_id: str) -> str:
84 | """
85 | View a user's current portfolio allocation
86 |
87 | Args:
88 | user_id: Unique identifier for the user
89 | """
90 | portfolio = load_portfolio(user_id)
91 |
92 | if not portfolio["stocks"] and not portfolio["bonds"]:
93 | return "Portfolio is empty. Use update_portfolio tool to add investments."
94 |
95 | result = ["# Current Portfolio Allocation", ""]
96 |
97 | # Add stocks
98 | if portfolio["stocks"]:
99 | result.append("## Stocks")
100 | for symbol, allocation in portfolio["stocks"].items():
101 | result.append(f"- {symbol}: {allocation}%")
102 | result.append("")
103 |
104 | # Add bonds
105 | if portfolio["bonds"]:
106 | result.append("## Bonds")
107 | for bond_id, allocation in portfolio["bonds"].items():
108 | result.append(f"- {bond_id}: {allocation}%")
109 | result.append("")
110 |
111 | # Add totals
112 | stock_allocation = sum(portfolio["stocks"].values())
113 | bond_allocation = sum(portfolio["bonds"].values())
114 | total_allocation = stock_allocation + bond_allocation
115 |
116 | result.append(f"## Summary")
117 | result.append(f"- Total stock allocation: {stock_allocation}%")
118 | result.append(f"- Total bond allocation: {bond_allocation}%")
119 | result.append(f"- Total allocation: {total_allocation}%")
120 |
121 | return "\n".join(result)
122 |
123 | @mcp_server.resource("portfolio://{user_id}")
124 | def get_portfolio_resource(user_id: str) -> str:
125 | """
126 | Get the current portfolio data as a resource
127 |
128 | Args:
129 | user_id: Unique identifier for the user
130 | """
131 | portfolio = load_portfolio(user_id)
132 | return json.dumps(portfolio, indent=2)
133 |
134 | if __name__ == "__main__":
135 | print("Portfolio Manager MCP Server starting with simplified configuration...", file=sys.stderr)
136 | mcp_server.run()
137 |
```
--------------------------------------------------------------------------------
/portfolio_server/tools/analysis_tools.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tools for analyzing portfolio data.
3 | """
4 | import json
5 | from typing import Dict
6 |
7 | from portfolio_server.data.storage import load_portfolio
8 | from portfolio_server.tools.stock_tools import get_stock_prices
9 |
10 | async def generate_portfolio_report(user_id: str) -> str:
11 | """
12 | Generate a comprehensive report on the current portfolio
13 |
14 | Args:
15 | user_id: Unique identifier for the user
16 | """
17 | portfolio = load_portfolio(user_id)
18 |
19 | if not portfolio["stocks"] and not portfolio["bonds"]:
20 | return "Portfolio is empty. Use update_portfolio tool to add investments."
21 |
22 | # Get stock price data
23 | stock_symbols = list(portfolio["stocks"].keys())
24 | price_data = json.loads(await get_stock_prices(stock_symbols))
25 |
26 | # Create a report
27 | report = ["# Portfolio Analysis Report", ""]
28 | report.append(f"## Current Allocation")
29 | report.append(f"- **Stocks**: {sum(portfolio['stocks'].values())}%")
30 | report.append(f"- **Bonds**: {sum(portfolio['bonds'].values())}%")
31 | report.append("")
32 |
33 | # Add performance section
34 | report.append("## Recent Performance")
35 |
36 | # Stock performance
37 | if stock_symbols:
38 | report.append("### Stocks")
39 | for symbol, allocation in portfolio["stocks"].items():
40 | if symbol in price_data and "percent_change" in price_data[symbol]:
41 | change = price_data[symbol]["percent_change"]
42 | contribution = (change * allocation) / 100
43 | report.append(f"- **{symbol}** ({allocation}% of portfolio): {change}% change, contributing {contribution:.2f}% to portfolio")
44 | else:
45 | report.append(f"- **{symbol}** ({allocation}% of portfolio): No recent data available")
46 | report.append("")
47 |
48 | # Bond performance (simplified as bonds have more complex data sources)
49 | if portfolio["bonds"]:
50 | report.append("### Bonds")
51 | report.append("Bond data typically changes less frequently than stocks.")
52 | for bond_id, allocation in portfolio["bonds"].items():
53 | report.append(f"- **{bond_id}** ({allocation}% of portfolio)")
54 | report.append("")
55 |
56 | # Add overall portfolio performance calculation
57 | # This is simplified but would be more comprehensive in a real app
58 | total_contribution = 0
59 | for symbol, allocation in portfolio["stocks"].items():
60 | if symbol in price_data and "percent_change" in price_data[symbol]:
61 | change = price_data[symbol]["percent_change"]
62 | contribution = (change * allocation) / 100
63 | total_contribution += contribution
64 |
65 | report.append(f"## Overall Portfolio Performance")
66 | report.append(f"The portfolio has changed approximately {total_contribution:.2f}% recently based on stock performance.")
67 |
68 | return "\n".join(report)
69 |
70 | async def get_investment_recommendations(user_id: str) -> str:
71 | """
72 | Get personalized investment recommendations based on current portfolio
73 |
74 | Args:
75 | user_id: Unique identifier for the user
76 | """
77 | portfolio = load_portfolio(user_id)
78 |
79 | if not portfolio["stocks"] and not portfolio["bonds"]:
80 | return "Portfolio is empty. Use update_portfolio tool to add investments first."
81 |
82 | # Calculate current asset allocation
83 | stock_allocation = sum(portfolio["stocks"].values())
84 | bond_allocation = sum(portfolio["bonds"].values())
85 | total_allocation = stock_allocation + bond_allocation
86 |
87 | recommendations = ["# Investment Recommendations", ""]
88 |
89 | # Check portfolio diversification
90 | stock_count = len(portfolio["stocks"])
91 | if stock_count < 5 and stock_allocation > 30:
92 | recommendations.append("## Diversification")
93 | recommendations.append("Your stock portfolio appears concentrated in a small number of stocks.")
94 | recommendations.append("Consider adding more stocks to reduce company-specific risk.")
95 | recommendations.append("")
96 |
97 | # Check asset allocation
98 | recommendations.append("## Asset Allocation")
99 | if stock_allocation > 0:
100 | stock_percent = (stock_allocation / total_allocation) * 100
101 | recommendations.append(f"Current allocation: {stock_percent:.1f}% stocks, {100-stock_percent:.1f}% bonds")
102 |
103 | # Very simplified recommendation based on stock/bond ratio
104 | # A real application would consider age, goals, risk tolerance, etc.
105 | if stock_percent > 80:
106 | recommendations.append("Your portfolio is heavily weighted toward stocks, which increases volatility.")
107 | recommendations.append("Consider increasing bond allocation for more stability.")
108 | elif stock_percent < 30:
109 | recommendations.append("Your portfolio is very conservative with a high bond allocation.")
110 | recommendations.append("Consider increasing stock allocation for greater long-term growth potential.")
111 | else:
112 | recommendations.append("Your current stock/bond allocation appears reasonably balanced.")
113 | recommendations.append("")
114 |
115 | # Check for overconcentration in individual positions
116 | for symbol, allocation in portfolio["stocks"].items():
117 | if allocation > 15: # Simplified threshold
118 | recommendations.append(f"**{symbol}** represents {allocation}% of your portfolio, which is relatively high.")
119 | recommendations.append(f"Consider reducing this position to limit single-stock risk.")
120 | recommendations.append("")
121 |
122 | if len(recommendations) <= 3: # Only has the title and asset allocation
123 | recommendations.append("Your portfolio appears well-structured based on basic checks.")
124 | recommendations.append("For more detailed recommendations, consider adding more information about your financial goals and risk tolerance.")
125 |
126 | return "\n".join(recommendations)
127 |
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | import sys
2 | import signal
3 | import logging
4 | import time
5 | import socket
6 | from portfolio_server.server import create_mcp_server
7 |
8 | # Configure logging
9 | logging.basicConfig(
10 | level=logging.INFO,
11 | format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
12 | )
13 | logger = logging.getLogger("portfolio_mcp")
14 |
15 | # Valid transport types
16 | VALID_TRANSPORTS = ["stdio", "sse", "http"]
17 |
18 | # Transport-specific configuration
19 | TRANSPORT_CONFIG = {
20 | "stdio": {
21 | "startup_delay": 0.1, # Short delay for stdio
22 | "max_retries": 1, # No retries needed for stdio
23 | "retry_delay": 0.5, # Delay between retries
24 | },
25 | "sse": {
26 | "startup_delay": 3.0, # Increased delay for SSE to establish connection
27 | "max_retries": 5, # Increased retries for network transports
28 | "retry_delay": 3.0, # Increased delay between retries
29 | "port": 8080, # Default SSE port
30 | "connection_timeout": 10.0 # Added timeout for connection attempts
31 | },
32 | "http": {
33 | "startup_delay": 0.5,
34 | "max_retries": 3,
35 | "retry_delay": 1.0,
36 | "port": 8000, # Default HTTP port
37 | }
38 | }
39 |
40 | # Global flag for shutdown
41 | is_shutting_down = False
42 |
43 | # Signal handlers for graceful shutdown
44 | def handle_shutdown_signal(signum, frame):
45 | global is_shutting_down
46 | if is_shutting_down:
47 | logger.warning("Forced shutdown triggered")
48 | sys.exit(1)
49 |
50 | logger.info(f"Shutdown signal received ({signum}), stopping server gracefully...")
51 | is_shutting_down = True
52 | # Note: The actual shutdown happens in the main loop
53 |
54 | # Register signal handlers
55 | signal.signal(signal.SIGINT, handle_shutdown_signal)
56 | signal.signal(signal.SIGTERM, handle_shutdown_signal)
57 |
58 | # Create the MCP server at module level
59 | try:
60 | mcp = create_mcp_server()
61 | logger.info("MCP server created successfully")
62 | except Exception as e:
63 | logger.error(f"Failed to create MCP server: {e}")
64 | sys.exit(1)
65 |
66 | def validate_transport(transport_type):
67 | """Validate that the transport type is supported."""
68 | if transport_type not in VALID_TRANSPORTS:
69 | logger.error(f"Invalid transport type: {transport_type}. Valid options are: {', '.join(VALID_TRANSPORTS)}")
70 | return False
71 | return True
72 |
73 |
74 | def check_connection_status(transport_type, config=None):
75 | """Check the connection status for the specified transport."""
76 | if config is None:
77 | config = {}
78 |
79 | if transport_type == "stdio":
80 | # stdio should always be available
81 | return True
82 | elif transport_type == "sse" or transport_type == "http":
83 | # Check if the port is available for sse/http
84 | port = config.get("port", TRANSPORT_CONFIG[transport_type]["port"])
85 | try:
86 | # Try to bind to the port to see if it's available
87 | sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
88 | sock.settimeout(1)
89 | result = sock.connect_ex(('127.0.0.1', port))
90 | sock.close()
91 |
92 | # If the connection was refused, the port is available (good)
93 | # If we could connect, someone else is using it (bad for our server)
94 | return result != 0
95 | except Exception as e:
96 | logger.warning(f"Error checking connection status for {transport_type}: {e}")
97 | return False
98 | else:
99 | logger.warning(f"Unknown transport type for status check: {transport_type}")
100 | return False
101 |
102 |
103 | def run_server_with_retry(transport_type, config=None):
104 | """Run the server with retry logic if the connection fails."""
105 | if config is None:
106 | config = {}
107 |
108 | # Get transport configuration
109 | transport_config = TRANSPORT_CONFIG.get(transport_type, {})
110 | max_retries = config.get("max_retries", transport_config.get("max_retries", 1))
111 | retry_delay = config.get("retry_delay", transport_config.get("retry_delay", 1.0))
112 | startup_delay = config.get("startup_delay", transport_config.get("startup_delay", 0.5))
113 |
114 | # Wait before starting to allow connections to establish
115 | logger.info(f"Waiting {startup_delay}s before starting server...")
116 | time.sleep(startup_delay)
117 |
118 | # Check connection status
119 | if not check_connection_status(transport_type, config):
120 | logger.warning(f"Transport {transport_type} may not be available. Proceeding with caution.")
121 |
122 | # Initialize retry counter
123 | retries = 0
124 | last_error = None
125 |
126 | while retries <= max_retries:
127 | try:
128 | # Create transport options
129 | transport_options = {}
130 |
131 | # Add transport-specific options
132 | if transport_type == "sse" or transport_type == "http":
133 | port = config.get("port", transport_config.get("port"))
134 | transport_options["port"] = port
135 |
136 | # Run the server with the configured transport and options
137 | logger.info(f"Starting server with {transport_type} transport (attempt {retries + 1}/{max_retries + 1})...")
138 | mcp.run(transport=transport_type, **transport_options)
139 |
140 | # If we get here, the server started successfully
141 | return True
142 | except ConnectionError as e:
143 | # Specific handling for connection errors
144 | last_error = e
145 | logger.error(f"Connection error starting server: {e}")
146 | except OSError as e:
147 | # Handle OS errors like port already in use
148 | last_error = e
149 | logger.error(f"OS error starting server: {e}")
150 | except Exception as e:
151 | # Catch any other exceptions
152 | last_error = e
153 | logger.error(f"Error starting server: {e}")
154 |
155 | # Increment retry counter
156 | retries += 1
157 |
158 | if retries <= max_retries:
159 | logger.info(f"Retrying in {retry_delay} seconds...")
160 | time.sleep(retry_delay)
161 |
162 | # If we get here, all retries failed
163 | logger.error(f"Failed to start server after {max_retries + 1} attempts. Last error: {last_error}")
164 | return False
165 |
166 | if __name__ == "__main__":
167 | logger.info("Portfolio Manager MCP Server starting")
168 |
169 | # Default
170 | transport = "stdio"
171 | config = {}
172 |
173 | # Check command line arguments for transport type
174 | if len(sys.argv) > 1:
175 | for arg in sys.argv[1:]:
176 | if arg == "--sse":
177 | transport = "sse"
178 | elif arg.startswith("--transport="):
179 | transport = arg.split("=")[1]
180 | elif arg.startswith("--port="):
181 | try:
182 | config["port"] = int(arg.split("=")[1])
183 | except ValueError:
184 | logger.error(f"Invalid port number: {arg.split('=')[1]}")
185 | sys.exit(1)
186 | elif arg.startswith("--retry="):
187 | try:
188 | config["max_retries"] = int(arg.split("=")[1])
189 | except ValueError:
190 | logger.error(f"Invalid retry count: {arg.split('=')[1]}")
191 | sys.exit(1)
192 |
193 | # Validate transport type
194 | if not validate_transport(transport):
195 | sys.exit(1)
196 |
197 | logger.info(f"Starting MCP server with transport: {transport}")
198 | if config:
199 | logger.info(f"Transport configuration: {config}")
200 |
201 | try:
202 | # For SSE transport, use the dedicated SSE module
203 | if transport == "sse":
204 | from portfolio_server.sse import run_sse_server
205 | port = config.get("port", TRANSPORT_CONFIG["sse"]["port"])
206 | host = "0.0.0.0" # Default host
207 | logger.info(f"Starting SSE server on {host}:{port}")
208 | run_sse_server(port=port, host=host)
209 | else:
210 | # For other transports, use the standard retry logic
211 | success = run_server_with_retry(transport, config)
212 | if not success:
213 | logger.error("Failed to start the server after all attempts.")
214 | sys.exit(1)
215 | except KeyboardInterrupt:
216 | # Handle keyboard interrupt (Ctrl+C)
217 | logger.info("Server interrupted via keyboard")
218 | except Exception as e:
219 | # Handle any other exceptions
220 | logger.error(f"Error running MCP server: {e}")
221 | sys.exit(1)
222 | finally:
223 | # Perform cleanup
224 | logger.info("Shutting down MCP server...")
225 | # Add any necessary cleanup code here
226 | logger.info("MCP server shutdown complete")
227 |
```