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