This is page 1 of 7. Use http://codebase.md/derekrbreese/fantasy-football-mcp-public?page={x} to view the full context.
# Directory Structure
```
├── .dockerignore
├── .env.example
├── .gitignore
├── config
│ ├── __init__.py
│ └── settings.py
├── Dockerfile
├── docs
│ ├── BYE_WEEKS_FIX.md
│ ├── LIVE_API_TEST_RESULTS.md
│ ├── LIVE_API_TESTING_SUMMARY.md
│ ├── PHASE_2B_REFACTOR_SUMMARY.md
│ ├── PROMPTS_AND_RESOURCES.md
│ ├── TEST_SUITE_SUMMARY.md
│ └── WAIVER_WIRE_VALIDATION_FIX.md
├── examples
│ ├── demo_enhancement_layer.py
│ ├── demos
│ │ ├── demo_consolidated_roster.py
│ │ ├── demo_ff_get_roster.py
│ │ └── demo_ff_get_waiver_wire.py
│ ├── example_client_llm_usage.py
│ └── hybrid_optimizer_example.py
├── fantasy_football_multi_league.py
├── fastmcp_server.py
├── INSTALLATION.md
├── LICENSE
├── lineup_optimizer.py
├── matchup_analyzer.py
├── position_normalizer.py
├── pyproject.toml
├── README.md
├── render.yaml
├── requirements.txt
├── sleeper_api.py
├── src
│ ├── __init__.py
│ ├── agents
│ │ ├── __init__.py
│ │ ├── cache_manager.py
│ │ ├── config.py
│ │ ├── data_fetcher.py
│ │ ├── decision.py
│ │ ├── draft_evaluator.py
│ │ ├── hybrid_optimizer.py
│ │ ├── integration.py
│ │ ├── llm_enhancement.py
│ │ ├── optimization.py
│ │ ├── reddit_analyzer.py
│ │ ├── roster_detector.py
│ │ ├── statistical.py
│ │ ├── user_interaction_engine.py
│ │ └── yahoo_auth.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── yahoo_client.py
│ │ └── yahoo_utils.py
│ ├── data
│ │ └── bye_weeks_2025.json
│ ├── handlers
│ │ ├── __init__.py
│ │ ├── admin_handlers.py
│ │ ├── analytics_handlers.py
│ │ ├── draft_handlers.py
│ │ ├── league_handlers.py
│ │ ├── matchup_handlers.py
│ │ ├── player_handlers.py
│ │ └── roster_handlers.py
│ ├── lineup_optimizer.py
│ ├── matchup_analyzer.py
│ ├── mcp_server.py
│ ├── models
│ │ ├── __init__.py
│ │ ├── draft.py
│ │ ├── lineup.py
│ │ ├── matchup.py
│ │ └── player.py
│ ├── parsers
│ │ ├── __init__.py
│ │ └── yahoo_parsers.py
│ ├── position_normalizer.py
│ ├── services
│ │ ├── __init__.py
│ │ ├── player_enhancement.py
│ │ └── reddit_service.py
│ ├── sleeper_api.py
│ ├── strategies
│ │ ├── __init__.py
│ │ ├── aggressive.py
│ │ ├── balanced.py
│ │ ├── base.py
│ │ └── conservative.py
│ ├── utils
│ │ ├── __init__.py
│ │ ├── bye_weeks.py
│ │ ├── constants.py
│ │ ├── roster_configs.py
│ │ └── scoring.py
│ └── yahoo_api_utils.py
├── tests
│ ├── conftest.py
│ ├── integration
│ │ ├── __init__.py
│ │ └── test_mcp_tools.py
│ ├── README.md
│ ├── TEST_CLEANUP_PLAN.md
│ ├── test_enhancement_layer.py
│ ├── test_live_api.py
│ ├── test_real_data.py
│ └── unit
│ ├── __init__.py
│ ├── test_api_client.py
│ ├── test_bye_weeks_utility.py
│ ├── test_bye_weeks.py
│ ├── test_handlers.py
│ ├── test_lineup_optimizer.py
│ └── test_parsers.py
├── utils
│ ├── reauth_yahoo.py
│ ├── refresh_yahoo_token.py
│ ├── setup_yahoo_auth.py
│ └── verify_setup.py
└── yahoo_api_utils.py
```
# Files
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
# Version control
.git
.gitignore
# Python
__pycache__
*.pyc
*.pyo
*.pyd
.Python
*.egg-info
dist
build
venv/
.venv/
env/
.env
.env.*
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Documentation
*.md
docs/
# Testing
.pytest_cache/
.coverage
htmlcov/
.tox/
# Logs
*.log
logs/
# Temporary files
tmp/
temp/
.tmp
# Development files
Dockerfile
.dockerignore
docker-compose.yml
# Local development scripts
test_*.py
*_test.py
# Cache directories
.cache/
.mypy_cache/
.ruff_cache/
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# Yahoo Fantasy Sports API Credentials
# Get these from https://developer.yahoo.com/apps/
YAHOO_CONSUMER_KEY=your_consumer_key_here
YAHOO_CONSUMER_SECRET=your_consumer_secret_here
YAHOO_ACCESS_TOKEN=your_access_token_here
YAHOO_REFRESH_TOKEN=your_refresh_token_here
# Yahoo User GUID (required for multi-league support)
# Get from https://api.login.yahoo.com/openid/v1/userinfo after OAuth
YAHOO_GUID=your_yahoo_guid_here
# Server Configuration
HOST=0.0.0.0
PORT=8000
# Logging
LOG_LEVEL=INFO
LOG_FILE=./logs/fantasy_football.log
# MCP Server Configuration
MCP_SERVER_NAME=fantasy-football
MCP_SERVER_VERSION=1.0.0
# Parallel Processing
MAX_WORKERS=10
ASYNC_TIMEOUT_SECONDS=30
# Feature Flags
ENABLE_ADVANCED_STATS=true
ENABLE_WEATHER_DATA=true
ENABLE_INJURY_REPORTS=true
# Reddit API Credentials (for sentiment analysis)
REDDIT_CLIENT_ID=your_reddit_client_id
REDDIT_CLIENT_SECRET=your_reddit_client_secret
REDDIT_USERNAME=your_reddit_username
# Yahoo OAuth Settings
YAHOO_REDIRECT_URI=http://localhost:8000/callback
YAHOO_CALLBACK_PORT=8000
YAHOO_CALLBACK_HOST=localhost
YAHOO_TOKEN_TIME=0
YAHOO_TOKEN_TYPE=bearer
# Remote MCP Server Configuration (for Render deployment)
DEBUG=true
ALLOWED_CLIENT_IDS=*
ALLOWED_REDIRECT_URIS=*
OAUTH_CLIENT_SECRET=secure-secret-change-this-in-production
CORS_ORIGINS=*
RENDER_EXTERNAL_URL=https://your-app-name.onrender.com
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Environment variables
.env
.env.*
.env.local
.env.*.local
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/
ENV/
build/
dist/
*.egg-info/
.pytest_cache/
# Tests (excluded from public repository)
tests/
test_*.py
*_test.py
# IDE
.vscode/
.idea/
.qoder/
.kilocode/
*.swp
*.swo
*~
# Cache and logs
.cache/
logs/
*.log
# Coverage reports
htmlcov/
.coverage
coverage.xml
*.cover
# Auth and tokens
.auth/
token.json
yahoo_token.json
*.token
# Development and debug files
debug_*.py
*_debug.py
find_*.py
parse_*.py
fetch_*.py
get_*.py
simple_*.py
standalone_*.py
mcp_server_test*.py
exchange_code.py
quick_validation.py
*_enhancement.py
*_enhanced.py
check_deployment.py
enhanced_mcp_*.py
# But allow production enhancement layer and official tests
!src/services/player_enhancement.py
!test_enhancement_layer.py
!test_real_data.py
!demo_enhancement_layer.py
# Temporary and analysis files
temp/
*_analysis.py
*_summary.py
cleanup_*.py
# JSON test data
all_leagues_raw.json
find_my_team.json
test_standings.json
yahoo_teams_raw.json
*_raw.json
# Config files with sensitive data
claude_desktop_config.json
mcp_config*.json
# Temporary files
*.tmp
*.bak
*.old
tmp/
# Development documentation
CLAUDE.md
CONTRIBUTING.md
FIX_*.md
CACHE_MANAGER_README.md
CLAUDE_AI_CONNECTION.md
REDDIT_SETUP.md
YAHOO_SETUP.md
ROSTER_CONSOLIDATION.md
# Deployment scripts
deploy_to_*.sh
run_local_mcp.sh
# macOS
.DS_Store
# Windows
Thumbs.db
ehthumbs.db
# Personal/development directories
.claude/
.qoder/
.devcontainer/
# Git hooks and personal scripts
.githooks/
hooks/
# Personal notes and private docs
PERSONAL_*.md
PRIVATE_*.md
TODO_*.md
# Runtime and deployment artifacts
*.pid
*.lock
deployment_log.txt
nohup.out
# OAuth and session files
oauth_state.json
session_*.json
callback_*.html
```
--------------------------------------------------------------------------------
/tests/README.md:
--------------------------------------------------------------------------------
```markdown
# Fantasy Football MCP Server - Test Suite
Comprehensive pytest test suite for ensuring code quality and reliability.
## Test Structure
```
tests/
├── conftest.py # Shared fixtures and test configuration
├── unit/ # Unit tests for individual modules
│ ├── test_api_client.py # Yahoo API client tests
│ ├── test_bye_weeks.py # Bye week integration tests (26 tests)
│ ├── test_bye_weeks_utility.py # Bye week utility unit tests (19 tests)
│ ├── test_handlers.py # MCP tool handler tests
│ ├── test_lineup_optimizer.py # Lineup optimization logic tests
│ └── test_parsers.py # Yahoo API response parser tests
└── integration/ # Integration tests for complete flows
└── test_mcp_tools.py # End-to-end MCP tool flow tests
```
## Running Tests
### Run All Tests
```bash
pytest tests/ -v
```
### Run Unit Tests Only
```bash
pytest tests/unit/ -v
```
### Run Integration Tests Only
```bash
pytest tests/integration/ -v
```
### Run with Coverage Report
```bash
pytest tests/ --cov=src --cov=lineup_optimizer --cov-report=term-missing --cov-report=html
```
### Run Specific Test File
```bash
pytest tests/unit/test_api_client.py -v
```
### Run Tests Matching Pattern
```bash
pytest tests/ -k "test_yahoo" -v
```
## Test Results Summary
**Total Tests**: 115 tests (107 unit + 8 integration)
**Pass Rate**: 100% ✅
### Bye Week Tests (45 tests)
- **test_bye_weeks_utility.py**: 19 unit tests for utility module
- **test_bye_weeks.py**: 26 integration tests for full system
- See [BYE_WEEKS_FIX.md](../BYE_WEEKS_FIX.md) for detailed documentation
### Coverage by Module
| Module | Statements | Coverage | Notes |
|--------|-----------|----------|-------|
| **src/api/yahoo_client.py** | 64 | **97%** | API client, token refresh, rate limiting |
| **src/handlers/admin_handlers.py** | 12 | **100%** | Admin MCP tool handlers |
| **src/handlers/league_handlers.py** | 64 | **86%** | League MCP tool handlers |
| **src/parsers/yahoo_parsers.py** | 125 | **95%** | Yahoo API response parsing (includes bye week validation) |
| **src/services/player_enhancement.py** | 85 | **100%** | Player enhancement and bye week detection |
| **src/utils/bye_weeks.py** | 42 | **100%** | Bye week utility module with static data fallback |
| **lineup_optimizer.py** | 312 | **51%** | Core optimization logic |
### Test Categories
#### Unit Tests (107 tests)
**API Client Tests** (`test_api_client.py`) - 12 tests
- ✅ Token management (get/set access tokens)
- ✅ Yahoo API calls with caching
- ✅ Rate limiting integration
- ✅ Automatic token refresh on 401 errors
- ✅ Error handling for API failures
- ✅ Cache hit/miss behavior
**Handler Tests** (`test_handlers.py`) - 9 tests
- ✅ Admin handlers (token refresh, API status, cache management)
- ✅ League handlers (get leagues, standings, teams)
- ✅ Error handling for missing parameters
- ✅ Response formatting
**Lineup Optimizer Tests** (`test_lineup_optimizer.py`) - 29 tests
- ✅ Utility functions (coerce_float, coerce_int, normalize_position)
- ✅ Match analytics tracking
- ✅ Player dataclass validation
- ✅ Roster parsing and validation
- ✅ Position normalization
- ✅ Invalid data handling
**Parser Tests** (`test_parsers.py`) - 12 tests
- ✅ Team roster parsing from Yahoo API responses
- ✅ Free agent/waiver wire parsing
- ✅ Handling malformed API responses
- ✅ Extracting player attributes (name, position, team, status)
- ✅ Ownership and injury data parsing
**Bye Week Utility Tests** (`test_bye_weeks_utility.py`) - 19 tests
- ✅ Static bye week data loading and caching
- ✅ Fallback logic (static data takes precedence over API)
- ✅ Team-to-bye-week mapping with API integration
- ✅ Cache management and reloading
- ✅ Error handling (file not found, invalid JSON, malformed data)
- ✅ All 32 NFL teams validation
- ✅ Real-world integration scenarios
**Bye Week Integration Tests** (`test_bye_weeks.py`) - 26 tests
- ✅ Bye week validation in player enhancement (11 tests)
- ✅ Yahoo parser bye week extraction (11 tests)
- ✅ Player context enhancement (2 tests)
- ✅ Main function integration (waiver wire, draft rankings) (2 tests)
- ✅ Full data flow with static data fallback
- ✅ Invalid data handling and range validation
#### Integration Tests (8 tests)
**MCP Tool Flows** (`test_mcp_tools.py`) - 8 tests
- ✅ Complete league tool workflow (discover → info → standings)
- ✅ Roster parsing to lineup optimizer pipeline
- ✅ Token refresh and API status workflow
- ✅ Yahoo API response transformation pipeline
- ✅ Error recovery in multi-stage pipelines
- ✅ Cache behavior (hits and misses)
## Test Fixtures
Shared fixtures in `conftest.py`:
- `mock_env_vars`: Mock Yahoo API credentials
- `mock_yahoo_league_response`: Sample Yahoo leagues API response
- `mock_yahoo_roster_response`: Sample Yahoo roster API response
- `mock_yahoo_free_agents_response`: Sample Yahoo free agents response
- `mock_yahoo_standings_response`: Sample Yahoo standings response
- `mock_rate_limiter`: Mock rate limiter for testing
- `mock_response_cache`: Mock response cache
- `sample_roster_data`: Sample parsed roster data
- `sample_sleeper_rankings`: Sample Sleeper API rankings
## Testing Best Practices
### For New Features
When adding new features:
1. **Write tests first** (TDD approach when possible)
2. **Test edge cases**: Empty responses, malformed data, network errors
3. **Test the happy path**: Ensure normal operations work correctly
4. **Mock external dependencies**: Yahoo API, Reddit API, etc.
5. **Aim for 80%+ coverage** on critical paths
### Test Naming Convention
```python
class TestModuleName:
def test_function_name_scenario(self):
"""Test description explaining what's being tested."""
# Arrange
# Act
# Assert
```
### Async Test Example
```python
@pytest.mark.asyncio
async def test_async_function(self):
"""Test async function behavior."""
result = await some_async_function()
assert result == expected
```
### Mock Yahoo API Example
```python
@pytest.mark.asyncio
async def test_with_mock_api(self, mock_yahoo_roster_response):
"""Test using mocked Yahoo API response."""
with patch("src.api.yahoo_client.yahoo_api_call") as mock_call:
mock_call.return_value = mock_yahoo_roster_response
result = await function_under_test()
assert result is not None
```
## Running Bye Week Tests
### Run All Bye Week Tests
```bash
pytest tests/unit/ -k "bye" -v
```
### Run Utility Tests Only
```bash
pytest tests/unit/test_bye_weeks_utility.py -v
```
### Run Integration Tests Only
```bash
pytest tests/unit/test_bye_weeks.py -v
```
### Run with Coverage
```bash
pytest tests/unit/ -k "bye" --cov=src.utils.bye_weeks --cov=src.parsers --cov=src.services -v
```
## Continuous Integration
Tests are designed to run in CI/CD pipelines:
```bash
# Install dependencies
pip install -r requirements.txt
# Run tests with coverage
pytest tests/ --cov=src --cov=lineup_optimizer --cov-report=xml
# Check coverage threshold (optional)
pytest tests/ --cov-fail-under=80
# Run bye week tests specifically
pytest tests/unit/ -k "bye" -v
```
## Adding New Tests
### 1. Create Test File
```python
"""Tests for new module."""
import pytest
from your_module import function_to_test
class TestNewFeature:
def test_basic_functionality(self):
"""Test basic feature works."""
result = function_to_test()
assert result == expected
```
### 2. Add Fixtures (if needed)
In `conftest.py`:
```python
@pytest.fixture
def sample_data():
"""Provide sample test data."""
return {"key": "value"}
```
### 3. Run New Tests
```bash
pytest tests/unit/test_new_file.py -v
```
## Troubleshooting
### Tests Fail with Import Errors
Ensure you're running from the project root:
```bash
cd /path/to/fantasy-football-mcp-server
pytest tests/
```
### Async Tests Not Running
Make sure `pytest-asyncio` is installed:
```bash
pip install pytest-asyncio
```
### Mock Issues
Use proper async mocking for aiohttp:
```python
# Create proper async context manager classes
class MockResponse:
status = 200
async def json(self):
return {"data": "value"}
class MockSession:
def get(self, *args, **kwargs):
class Context:
async def __aenter__(self):
return MockResponse()
async def __aexit__(self, *args):
return None
return Context()
```
## Future Test Improvements
- [ ] Add tests for draft evaluation algorithms
- [ ] Add tests for matchup analyzer
- [ ] Add tests for position normalizer
- [ ] Add tests for sleeper API integration
- [ ] Add performance/benchmark tests
- [ ] Add end-to-end tests with real API (optional, behind flag)
- [ ] Increase coverage for complex handlers (roster, draft, player)
- [ ] Add property-based testing with Hypothesis
## Test Markers
Available pytest markers:
- `@pytest.mark.unit`: Unit test
- `@pytest.mark.integration`: Integration test
- `@pytest.mark.slow`: Slow-running test
- `@pytest.mark.asyncio`: Async test
Filter by marker:
```bash
pytest tests/ -m "unit and not slow"
```
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Fantasy Football MCP Server
A comprehensive Model Context Protocol (MCP) server for Yahoo Fantasy Football that provides intelligent lineup optimization, draft assistance, and league management through AI-powered tools.
## 🚀 Features
### Core Capabilities
- **Multi-League Support** – Automatically discovers and manages all Yahoo Fantasy Football leagues associated with your account
- **🆕 Player Enhancement Layer** – Intelligent projection adjustments with bye week detection, recent performance stats, and breakout/declining player flags
- **Intelligent Lineup Optimization** – Advanced algorithms considering matchups, expert projections, and position-normalized value
- **Draft Assistant** – Real-time draft recommendations with strategy-based analysis and VORP calculations
- **Comprehensive Analytics** – Reddit sentiment analysis, team comparisons, and performance metrics
- **Multiple Deployment Options** – FastMCP, traditional MCP, Docker, and cloud deployment support
### Advanced Analytics
- **Position Normalization** – Smart FLEX decisions accounting for different position baselines
- **Multi-Source Projections** – Combines Yahoo and Sleeper expert rankings with matchup analysis
- **Strategy-Based Optimization** – Conservative, aggressive, and balanced approaches
- **Volatility Scoring** – Floor vs ceiling analysis for consistent or boom-bust plays
- **Live Draft Support** – Real-time recommendations during active drafts
## 🆕 Player Enhancement Layer
The enhancement layer enriches player data with real-world context to fix stale projections and prevent common mistakes:
### Key Features
✅ **Bye Week Detection** – Automatically zeros projections and displays "BYE WEEK - DO NOT START" for players on bye, preventing accidental starts
✅ **Recent Performance Stats** – Fetches last 1-3 weeks of actual performance from Sleeper API and displays trends (L3W avg: X.X pts/game)
✅ **Performance Flags** – Intelligent alerts including:
- `BREAKOUT_CANDIDATE` – Recent performance > 150% of projection
- `TRENDING_UP` – Recent performance exceeds projection
- `DECLINING_ROLE` – Recent performance < 70% of projection
- `HIGH_CEILING` – Explosive upside potential
- `CONSISTENT` – Reliable, steady performance
✅ **Adjusted Projections** – Blends recent reality with stale projections for more accurate start/sit decisions (60/40 or 70/30 weighting based on confidence)
### Example
**Before Enhancement:**
```json
{
"name": "Rico Dowdle",
"sleeper_projection": 4.0,
"recommendation": "Bench"
}
```
**After Enhancement:**
```json
{
"name": "Rico Dowdle",
"sleeper_projection": 4.0,
"adjusted_projection": 14.8,
"performance_flags": ["BREAKOUT_CANDIDATE", "TRENDING_UP"],
"enhancement_context": "Recent breakout: averaging 18.5 pts over last 3 weeks",
"recommendation": "Strong Start"
}
```
The enhancement layer is **non-breaking** and automatically applies to:
- `ff_get_roster` (with `include_external_data=True`)
- `ff_get_waiver_wire` (with `include_external_data=True`)
- `ff_get_players` (with `include_external_data=True`)
- `ff_build_lineup` (automatic)
## 🛠️ Available MCP Tools
### League & Team Management
- `ff_get_leagues` – List all leagues for your authenticated Yahoo account
- `ff_get_league_info` – Retrieve detailed league metadata and team information
- `ff_get_standings` – View current league standings with wins, losses, and points
- `ff_get_roster` – Inspect detailed roster information for any team
- `ff_get_matchup` – Analyze weekly matchup details and projections
- `ff_compare_teams` – Side-by-side team roster comparisons for trades/analysis
- `ff_build_lineup` – Generate optimal lineups using advanced optimization algorithms
### Player Discovery & Waiver Wire
- `ff_get_players` – Browse available free agents with ownership percentages
- `ff_get_waiver_wire` – Smart waiver wire targets with expert analysis (configurable count)
- `ff_get_draft_rankings` – Access Yahoo's pre-draft rankings and ADP data
### Draft Assistant Tools
- `ff_get_draft_recommendation` – AI-powered draft pick suggestions with strategy analysis
- `ff_analyze_draft_state` – Real-time roster needs and positional analysis during drafts
- `ff_get_draft_results` – Post-draft analysis with grades and team summaries
### Advanced Analytics
- `ff_analyze_reddit_sentiment` – Social media sentiment analysis for player buzz and injury updates
- `ff_get_api_status` – Monitor cache performance and Yahoo API rate limiting
- `ff_clear_cache` – Clear cached responses for fresh data (with pattern support)
- `ff_refresh_token` – Automatically refresh Yahoo OAuth tokens
## 📦 Installation
### Quick Start
```bash
git clone https://github.com/derekrbreese/fantasy-football-mcp-public.git
cd fantasy-football-mcp-public
pip install -r requirements.txt
```
### Yahoo API Setup
1. Create a Yahoo Developer App at [developer.yahoo.com](https://developer.yahoo.com)
2. Note your Consumer Key and Consumer Secret
3. Complete OAuth flow using included scripts
## ⚙️ Configuration
Create a `.env` file with your Yahoo API credentials:
```env
YAHOO_CONSUMER_KEY=your_consumer_key_here
YAHOO_CONSUMER_SECRET=your_consumer_secret_here
YAHOO_ACCESS_TOKEN=your_access_token
YAHOO_REFRESH_TOKEN=your_refresh_token
YAHOO_GUID=your_yahoo_guid
```
### Initial Authentication
```bash
# First-time setup
python setup_yahoo_auth.py
# Or manual authentication
python reauth_yahoo.py
```
## 🚀 Deployment Options
### Local Development (FastMCP)
```bash
python fastmcp_server.py
```
Connect via HTTP transport at `http://localhost:8000`
### Claude Code Integration (Stdio)
```bash
python fantasy_football_multi_league.py
```
### Docker Deployment
```bash
docker build -t fantasy-football-mcp .
docker run -p 8080:8080 --env-file .env fantasy-football-mcp
```
### Cloud Deployment (Render/Railway/etc.)
The server includes multiple compatibility layers for various cloud platforms:
- `render_server.py` - Render.com deployment
- `simple_mcp_server.py` - Generic HTTP/WebSocket server
- `fastmcp_server.py` - FastMCP cloud deployments
## 🧪 Testing
```bash
# Run full test suite
pytest
# Test OAuth authentication
python tests/test_oauth.py
# Test MCP connection
python tests/test_mcp_client.py
```
## 📁 Project Structure
```
fantasy-football-mcp-public/
├── fastmcp_server.py # FastMCP HTTP server implementation
├── fantasy_football_multi_league.py # Main MCP stdio server
├── lineup_optimizer.py # Advanced lineup optimization engine
├── matchup_analyzer.py # Defensive matchup analysis
├── position_normalizer.py # FLEX position value calculations
├── src/
│ ├── agents/ # Specialized analysis agents
│ ├── models/ # Data models for players, lineups, drafts
│ ├── strategies/ # Draft and lineup strategies
│ ├── services/ # Player enhancement and external integrations
│ └── utils/ # Utility functions and configurations
├── tests/ # Comprehensive test suite
├── utils/ # Authentication and token management
└── requirements.txt # Python dependencies
```
## 🔧 Advanced Configuration
### Strategy Weights (Balanced Default)
```python
{
"yahoo": 0.40, # Yahoo expert projections
"sleeper": 0.40, # Sleeper expert rankings
"matchup": 0.10, # Defensive matchup analysis
"trending": 0.05, # Player trending data
"momentum": 0.05 # Recent performance
}
```
### Draft Strategies
- **Conservative**: Prioritize proven players, minimize risk
- **Aggressive**: Target high-upside breakout candidates
- **Balanced**: Optimal mix of safety and ceiling potential
### Position Scoring Baselines
- RB: ~11 points (standard scoring)
- WR: ~10 points (standard scoring)
- TE: ~7 points (standard scoring)
- FLEX calculations include position scarcity adjustments
## 📊 Performance Metrics
The optimization engine targets:
- **85%+** accuracy on start/sit decisions
- **+2.0** points per optimal decision on average
- **90%+** lineup efficiency vs. manual selection
- **Position-normalized FLEX** decisions to avoid TE traps
## 🔍 Troubleshooting
### Common Issues
**Authentication Errors**
```bash
# Refresh expired tokens (expire hourly)
python utils/refresh_token.py
# Full re-authentication if refresh fails
python reauth_yahoo.py
```
**Only One League Showing**
- Verify `YAHOO_GUID` matches your Yahoo account
- Ensure leagues are active for current season
- Check team ownership detection in logs
**Rate Limiting**
- Yahoo allows 1000 requests/hour
- Server implements 900/hour safety limit
- Use `ff_get_api_status` to monitor usage
- Clear cache with `ff_clear_cache` if needed
**Stale Data**
- Cache TTLs: Leagues (1hr), Standings (5min), Players (15min)
- Force refresh with `ff_clear_cache` tool
- Check last update times in `ff_get_api_status`
## 🤝 Contributing
This is the public version of the Fantasy Football MCP Server. For contributing:
1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request
## 📄 License
MIT License - see LICENSE file for details
## 🙏 Acknowledgments
- Yahoo Fantasy Sports API for comprehensive league data
- Sleeper API for expert rankings and defensive analysis
- Reddit API for player sentiment analysis
- Model Context Protocol (MCP) framework
---
**Note**: This server requires active Yahoo Fantasy Football leagues and valid API credentials. Ensure you have proper authorization before accessing league data.
```
--------------------------------------------------------------------------------
/config/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/models/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/utils/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/unit/__init__.py:
--------------------------------------------------------------------------------
```python
"""Unit tests for individual modules."""
```
--------------------------------------------------------------------------------
/tests/integration/__init__.py:
--------------------------------------------------------------------------------
```python
"""Integration tests for MCP tool flows."""
```
--------------------------------------------------------------------------------
/src/services/__init__.py:
--------------------------------------------------------------------------------
```python
"""Services for external integrations."""
from .reddit_service import analyze_reddit_sentiment
__all__ = ["analyze_reddit_sentiment"]
```
--------------------------------------------------------------------------------
/src/parsers/__init__.py:
--------------------------------------------------------------------------------
```python
"""Yahoo API response parsers."""
from .yahoo_parsers import parse_team_roster, parse_yahoo_free_agent_players
__all__ = ["parse_team_roster", "parse_yahoo_free_agent_players"]
```
--------------------------------------------------------------------------------
/src/api/__init__.py:
--------------------------------------------------------------------------------
```python
"""Yahoo API client module."""
from .yahoo_client import (
YAHOO_API_BASE,
get_access_token,
refresh_yahoo_token,
set_access_token,
yahoo_api_call,
)
__all__ = [
"yahoo_api_call",
"refresh_yahoo_token",
"get_access_token",
"set_access_token",
"YAHOO_API_BASE",
]
```
--------------------------------------------------------------------------------
/src/agents/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Fantasy Football MCP Agents Module.
This module provides intelligent agents for fantasy football analysis and decision making.
"""
from .decision import (
DecisionAgent,
RiskToleranceProfile,
DecisionFactor,
DecisionNode,
MultiCriteriaScore,
DecisionExplanation,
)
__all__ = [
"DecisionAgent",
"RiskToleranceProfile",
"DecisionFactor",
"DecisionNode",
"MultiCriteriaScore",
"DecisionExplanation",
]
```
--------------------------------------------------------------------------------
/src/data/bye_weeks_2025.json:
--------------------------------------------------------------------------------
```json
{
"ARI": 8,
"ATL": 5,
"BAL": 7,
"BUF": 7,
"CAR": 14,
"CHI": 5,
"CIN": 10,
"CLE": 9,
"DAL": 10,
"DEN": 12,
"DET": 8,
"GB": 5,
"HOU": 6,
"IND": 11,
"JAC": 8,
"KC": 10,
"LAC": 12,
"LAR": 8,
"LV": 8,
"MIA": 12,
"MIN": 6,
"NE": 14,
"NO": 11,
"NYG": 14,
"NYJ": 9,
"PHI": 9,
"PIT": 5,
"SEA": 8,
"SF": 14,
"TB": 9,
"TEN": 10,
"WAS": 12
}
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
# Core MCP server dependencies
fastmcp==2.12.3
mcp==1.14.0
yfpy==17.0.0
python-dotenv==1.1.1
pydantic==2.11.9
pydantic-settings==2.10.1
# Async and caching utilities
asyncio-pool==0.6.0
aiohttp==3.11.11
aiocache==0.12.3
# Data processing
pandas==2.3.2
numpy==2.3.3
# Logging and CLI helpers
loguru==0.7.3
click==8.2.1
rich==14.1.0
# Yahoo and Reddit integrations
praw==7.8.1
prawcore==2.4.0
textblob==0.19.0
requests==2.32.5
yahoo-oauth==2.1.1
stringcase==1.2.0
# Testing
pytest==8.4.2
pytest-asyncio==1.2.0
pytest-mock==3.15.1
pytest-cov==6.0.0 # Code coverage reporting
```
--------------------------------------------------------------------------------
/src/handlers/analytics_handlers.py:
--------------------------------------------------------------------------------
```python
"""Analytics MCP tool handlers."""
from src.services import analyze_reddit_sentiment
async def handle_ff_analyze_reddit_sentiment(arguments: dict) -> dict:
"""Analyze Reddit sentiment for specified players.
Args:
arguments: Dict containing:
- players: List of player names to analyze
- time_window_hours: Time window in hours (default: 48)
Returns:
Dict with sentiment analysis results
"""
players = arguments.get("players", [])
time_window = arguments.get("time_window_hours", 48)
if not players:
return {"error": "No players specified for sentiment analysis"}
return await analyze_reddit_sentiment(players, time_window)
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use Python 3.11 slim image for compatibility with dependencies
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Copy requirements first for better layer caching
COPY requirements.txt .
# Upgrade pip and install Python dependencies
RUN pip install --no-cache-dir --upgrade pip \
&& pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Create non-root user for security
RUN useradd --create-home --shell /bin/bash app \
&& chown -R app:app /app
USER app
# Set environment variables
ENV PYTHONPATH=/app
ENV PYTHONUNBUFFERED=1
# Expose port (Cloud Run will set PORT env var)
EXPOSE 8080
# Run FastMCP server directly (no_auth_server shim removed)
CMD python fastmcp_server.py
```
--------------------------------------------------------------------------------
/render.yaml:
--------------------------------------------------------------------------------
```yaml
services:
- type: web
name: fantasy-football-mcp-server
runtime: python
buildCommand: "pip install -r requirements.txt"
startCommand: "python fastmcp_server.py"
healthCheckPath: /health
numInstances: 1
envVars:
- key: PYTHON_VERSION
value: 3.11.0
- key: DEBUG
value: true
- key: ALLOWED_REDIRECT_URIS
value: "https://claude.ai/api/mcp/auth_callback,https://claude.ai/oauth/callback,https://claude.ai/*"
- key: ALLOWED_CLIENT_IDS
value: "Claude,claude-ai,claude.ai,anthropic,mcp-client,*"
- key: OAUTH_CLIENT_SECRET
generateValue: true
- key: YAHOO_CONSUMER_KEY
sync: false
- key: YAHOO_CONSUMER_SECRET
sync: false
- key: YAHOO_ACCESS_TOKEN
sync: false
- key: YAHOO_REFRESH_TOKEN
sync: false
- key: YAHOO_GUID
sync: false
- key: PORT
value: 8080
```
--------------------------------------------------------------------------------
/examples/demos/demo_ff_get_roster.py:
--------------------------------------------------------------------------------
```python
import asyncio
import json
from fantasy_football_multi_league import call_tool
async def main():
"""Demonstrate ff_get_roster tool output with full data level."""
# Discover leagues
leagues_response = await call_tool("ff_get_leagues", {})
leagues_payload = json.loads(leagues_response[0].text)
leagues = leagues_payload.get("leagues", [])
if not leagues:
print("No leagues found. Ensure Yahoo credentials are set.")
return
league_key = leagues[0].get("key")
print(f"Using league_key: {league_key}")
print(f"League name: {leagues[0].get('name')}")
# Call ff_get_roster with full data level
roster_args = {
"league_key": league_key,
"data_level": "full",
"include_projections": True,
"include_external_data": True,
"include_analysis": True,
}
roster_response = await call_tool("ff_get_roster", roster_args)
roster_result = json.loads(roster_response[0].text)
if roster_result.get("status") == "success":
print("\n=== ff_get_roster Output (Full Data Level) ===")
print(json.dumps(roster_result, indent=2))
else:
print(f"Error: {roster_result}")
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/src/handlers/admin_handlers.py:
--------------------------------------------------------------------------------
```python
"""Admin and system management handlers for MCP tools."""
from typing import Dict
from src.api import refresh_yahoo_token
from src.api.yahoo_utils import rate_limiter, response_cache
async def handle_ff_refresh_token(arguments: Dict) -> Dict:
"""Refresh the Yahoo OAuth access token.
Args:
arguments: Empty dict (no arguments required)
Returns:
Status dict with refresh result
"""
return await refresh_yahoo_token()
async def handle_ff_get_api_status(arguments: Dict) -> Dict:
"""Get current API rate limit and cache status.
Args:
arguments: Empty dict (no arguments required)
Returns:
Dict with rate_limit and cache statistics
"""
return {
"rate_limit": rate_limiter.get_status(),
"cache": response_cache.get_stats(),
}
async def handle_ff_clear_cache(arguments: Dict) -> Dict:
"""Clear the Yahoo API response cache.
Args:
arguments: Optional 'pattern' to clear specific cache entries
Returns:
Status dict confirming cache clear
"""
pattern = arguments.get("pattern")
await response_cache.clear(pattern)
suffix = f" for pattern: {pattern}" if pattern else " completely"
return {
"status": "success",
"message": f"Cache cleared{suffix}",
}
```
--------------------------------------------------------------------------------
/examples/demos/demo_ff_get_waiver_wire.py:
--------------------------------------------------------------------------------
```python
import asyncio
import json
from fantasy_football_multi_league import call_tool
async def main():
"""Demonstrate ff_get_waiver_wire tool output with full enhancements."""
# Discover leagues
leagues_response = await call_tool("ff_get_leagues", {})
leagues_payload = json.loads(leagues_response[0].text) # type: ignore
leagues = leagues_payload.get("leagues", [])
if not leagues:
print("No leagues found. Ensure Yahoo credentials are set.")
return
league_key = leagues[0].get("key")
print(f"Using league_key: {league_key}")
print(f"League name: {leagues[0].get('name')}")
# Call ff_get_waiver_wire with full enhancements
waiver_args = {
"league_key": league_key,
"position": "all",
"sort": "rank",
"include_expert_analysis": True,
"include_projections": True,
"include_external_data": True,
}
waiver_response = await call_tool("ff_get_waiver_wire", waiver_args)
waiver_result = json.loads(waiver_response[0].text) # type: ignore
if waiver_result.get("status") == "success":
print("\n=== ff_get_waiver_wire Output (Full Enhancements) ===")
print(json.dumps(waiver_result, indent=2))
else:
print(f"Error: {waiver_result}")
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/config/settings.py:
--------------------------------------------------------------------------------
```python
"""
Configuration settings for the Fantasy Football MCP server.
"""
import os
from pathlib import Path
from typing import Optional
from pydantic_settings import BaseSettings
from pydantic import Field
class Settings(BaseSettings):
"""Application settings loaded from environment variables."""
# Yahoo API Configuration
yahoo_client_id: str = Field(..., env="YAHOO_CLIENT_ID")
yahoo_client_secret: str = Field(..., env="YAHOO_CLIENT_SECRET")
# Cache Configuration
cache_dir: Path = Field(default=Path("./.cache"), env="CACHE_DIR")
cache_ttl_seconds: int = Field(default=3600, env="CACHE_TTL_SECONDS")
# API Rate Limiting
yahoo_api_rate_limit: int = Field(default=100, env="YAHOO_API_RATE_LIMIT")
yahoo_api_rate_window_seconds: int = Field(default=3600, env="YAHOO_API_RATE_WINDOW_SECONDS")
# Logging
log_level: str = Field(default="INFO", env="LOG_LEVEL")
log_file: Path = Field(default=Path("./logs/fantasy_football.log"), env="LOG_FILE")
# MCP Server Configuration
mcp_server_name: str = Field(default="fantasy-football", env="MCP_SERVER_NAME")
mcp_server_version: str = Field(default="1.0.0", env="MCP_SERVER_VERSION")
# Parallel Processing
max_workers: int = Field(default=10, env="MAX_WORKERS")
async_timeout_seconds: int = Field(default=30, env="ASYNC_TIMEOUT_SECONDS")
# Feature Flags
enable_advanced_stats: bool = Field(default=True, env="ENABLE_ADVANCED_STATS")
enable_weather_data: bool = Field(default=True, env="ENABLE_WEATHER_DATA")
enable_injury_reports: bool = Field(default=True, env="ENABLE_INJURY_REPORTS")
# Yahoo OAuth Configuration
yahoo_redirect_uri: str = Field(default="http://localhost:8090", env="YAHOO_REDIRECT_URI")
yahoo_callback_port: int = Field(default=8090, env="YAHOO_CALLBACK_PORT")
yahoo_callback_host: str = Field(default="localhost", env="YAHOO_CALLBACK_HOST")
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Ensure cache directory exists
self.cache_dir.mkdir(parents=True, exist_ok=True)
# Ensure log directory exists
self.log_file.parent.mkdir(parents=True, exist_ok=True)
```
--------------------------------------------------------------------------------
/docs/LIVE_API_TEST_RESULTS.md:
--------------------------------------------------------------------------------
```markdown
# Live API Test Results - Phase 2b Refactoring
**Test Date:** 2025-10-02 13:26:36
**Branch:** consolidate-fastmcp
**Python Version:** 3.12.1
## Executive Summary
- **Total Tests:** 22
- **✓ Passed:** 22
- **✗ Failed:** 0
- **Pass Rate:** 100.0%
- **Total Time:** 14.67s
- **API Calls:** 22
**Assessment:** ✅ **EXCELLENT** - Refactoring is solid, ready to merge
## Detailed Results by Category
### Admin Handlers (3/3 passed)
- **✓ ff_get_api_status** - 0.00s
- Tool: `ff_get_api_status`
- **✓ ff_clear_cache** - 0.00s
- Tool: `ff_clear_cache`
- **✓ ff_refresh_token** - 0.06s
- Tool: `ff_refresh_token`
### League Handlers (7/7 passed)
- **✓ Get leagues for context** - 0.22s
- Tool: `ff_get_leagues`
- **✓ Get teams for context** - 0.15s
- Tool: `ff_get_teams`
- **✓ ff_get_leagues** - 0.00s
- Tool: `ff_get_leagues`
- **✓ ff_get_league_info** - 0.19s
- Tool: `ff_get_league_info`
- **✓ ff_get_standings** - 0.18s
- Tool: `ff_get_standings`
- **✓ ff_get_teams** - 0.00s
- Tool: `ff_get_teams`
- **✓ ff_compare_teams** - 0.10s
- Tool: `ff_compare_teams`
### Roster Handlers (3/3 passed)
- **✓ ff_get_roster (basic)** - 0.09s
- Tool: `ff_get_roster`
- **✓ ff_get_roster (standard)** - 2.81s
- Tool: `ff_get_roster`
- **✓ ff_get_roster (full)** - 1.13s
- Tool: `ff_get_roster`
### Matchup Handlers (2/2 passed)
- **✓ ff_get_matchup** - 0.54s
- Tool: `ff_get_matchup`
- **✓ ff_build_lineup** - 0.90s
- Tool: `ff_build_lineup`
### Player Handlers (2/2 passed)
- **✓ ff_get_players (QB)** - 0.45s
- Tool: `ff_get_players`
- **✓ ff_get_waiver_wire** - 1.60s
- Tool: `ff_get_waiver_wire`
### Draft Handlers (4/4 passed)
- **✓ ff_get_draft_results** - 0.00s
- Tool: `ff_get_draft_results`
- **✓ ff_get_draft_rankings** - 0.16s
- Tool: `ff_get_draft_rankings`
- **✓ ff_get_draft_recommendation** - 0.54s
- Tool: `ff_get_draft_recommendation`
- **✓ ff_analyze_draft_state** - 0.00s
- Tool: `ff_analyze_draft_state`
### Analytics Handlers (1/1 passed)
- **✓ ff_analyze_reddit_sentiment** - 5.54s
- Tool: `ff_analyze_reddit_sentiment`
## Performance Metrics
- **Average Response Time:** 0.67s
- **Fastest Test:** 0.00s (ff_get_leagues)
- **Slowest Test:** 5.54s (ff_analyze_reddit_sentiment)
## Recommendations
✅ **Ready to merge** - All critical handlers working correctly
- Consider merging consolidate-fastmcp branch to main
- Update documentation with any new features
---
*Generated by `test_live_api.py` on 2025-10-02*
```
--------------------------------------------------------------------------------
/src/strategies/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Fantasy football lineup strategies module.
This module provides various lineup construction strategies for different
contest types and risk preferences. Each strategy implements a common
interface while optimizing for different objectives.
Available strategies:
- ConservativeStrategy: Floor-focused, risk-averse lineup construction
- AggressiveStrategy: Ceiling-focused, high-upside tournament strategy
- BalancedStrategy: Risk/reward optimized, versatile approach
Strategy Selection Guide:
- Cash games & 50/50s: ConservativeStrategy
- Large GPPs & tournaments: AggressiveStrategy
- Mixed contests & medium fields: BalancedStrategy
"""
from .base import BaseLineupStrategy, StrategyConfig, StrategyType, WeightAdjustment, PlayerScore
from .conservative import ConservativeStrategy
from .aggressive import AggressiveStrategy
from .balanced import BalancedStrategy
__all__ = [
# Base classes and types
"BaseLineupStrategy",
"StrategyConfig",
"StrategyType",
"WeightAdjustment",
"PlayerScore",
# Concrete strategy implementations
"ConservativeStrategy",
"AggressiveStrategy",
"BalancedStrategy",
]
# Strategy registry for dynamic instantiation
STRATEGY_REGISTRY = {
StrategyType.CONSERVATIVE: ConservativeStrategy,
StrategyType.AGGRESSIVE: AggressiveStrategy,
StrategyType.BALANCED: BalancedStrategy,
}
def get_strategy(strategy_type: StrategyType, config: StrategyConfig = None) -> BaseLineupStrategy:
"""
Factory function to create strategy instances.
Args:
strategy_type: The type of strategy to create
config: Optional strategy configuration
Returns:
Initialized strategy instance
Raises:
ValueError: If strategy_type is not supported
"""
if strategy_type not in STRATEGY_REGISTRY:
raise ValueError(f"Unsupported strategy type: {strategy_type}")
strategy_class = STRATEGY_REGISTRY[strategy_type]
return strategy_class(config=config)
def get_available_strategies() -> list[str]:
"""
Get list of available strategy names.
Returns:
List of strategy type names
"""
return [strategy_type.value for strategy_type in STRATEGY_REGISTRY.keys()]
def get_strategy_for_contest_type(contest_type: str) -> StrategyType:
"""
Get recommended strategy type for a contest type.
Args:
contest_type: Contest type string (e.g., "GPP", "Cash", "Tournament")
Returns:
Recommended strategy type
"""
contest_lower = contest_type.lower()
# GPP/Tournament -> Aggressive
if any(
keyword in contest_lower
for keyword in ["gpp", "tournament", "millionaire", "large", "high-stakes"]
):
return StrategyType.AGGRESSIVE
# Cash games -> Conservative
if any(
keyword in contest_lower for keyword in ["cash", "50/50", "double", "head-to-head", "safe"]
):
return StrategyType.CONSERVATIVE
# Default to balanced for mixed/unknown contest types
return StrategyType.BALANCED
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=65", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "fantasy-football-mcp"
version = "1.0.0"
description = "Production-grade MCP server for Yahoo Fantasy Sports with sophisticated lineup optimization"
readme = "README.md"
requires-python = ">=3.9"
license = {text = "MIT"}
authors = [
{name = "Your Name", email = "[email protected]"}
]
keywords = [
"fantasy-football",
"mcp",
"yahoo-fantasy",
"lineup-optimization",
"sports-analytics"
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Games/Entertainment"
]
dependencies = [
"mcp>=1.2.0",
"yfpy>=16.0.3",
"pydantic>=2.10.4",
"pydantic-settings>=2.0.0",
"python-dotenv>=1.0.1",
"asyncio-pool>=0.6.0",
"aiohttp>=3.11.11",
"aiocache>=0.12.3",
"pandas>=2.2.3",
"numpy>=1.26.4",
"loguru>=0.7.3",
"click>=8.1.8",
"rich>=13.9.6"
]
[project.optional-dependencies]
dev = [
"pytest>=8.3.4",
"pytest-asyncio>=0.25.2",
"pytest-mock>=3.14.0",
"pytest-cov>=6.0.0",
"black>=24.10.0",
"mypy>=1.14.1",
"ruff>=0.9.1",
"ipython>=8.18.0"
]
[project.scripts]
fantasy-football-mcp = "src.mcp_server:main"
[project.urls]
Homepage = "https://github.com/yourusername/fantasy-football-mcp"
Documentation = "https://github.com/yourusername/fantasy-football-mcp/wiki"
Repository = "https://github.com/yourusername/fantasy-football-mcp.git"
Issues = "https://github.com/yourusername/fantasy-football-mcp/issues"
[tool.setuptools]
packages = ["src"]
[tool.setuptools.package-data]
src = ["*.json", "*.yaml", "*.yml"]
[tool.black]
line-length = 100
target-version = ["py39", "py310", "py311", "py312"]
include = '\.pyi?$'
extend-exclude = '''
/(
\.git
| \.mypy_cache
| \.pytest_cache
| \.venv
| build
| dist
)/
'''
[tool.ruff]
line-length = 100
target-version = "py39"
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (handled by black)
"B008", # do not perform function calls in argument defaults
"W191", # indentation contains tabs
]
[tool.mypy]
python_version = "3.9"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_any_unimported = false
disallow_any_expr = false
disallow_any_decorated = false
disallow_any_explicit = false
disallow_any_generics = false
disallow_subclassing_any = false
no_implicit_optional = true
check_untyped_defs = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
follow_imports = "silent"
ignore_missing_imports = true
[tool.pytest.ini_options]
minversion = "6.0"
addopts = "-ra -q --strict-markers"
testpaths = ["tests"]
pythonpath = ["."]
asyncio_mode = "auto"
markers = [
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
"integration: marks tests as integration tests",
"unit: marks tests as unit tests",
"asyncio: marks coroutine tests (provided by pytest-asyncio)"
]
[tool.coverage.run]
source = ["src"]
omit = ["*/tests/*", "*/test_*.py"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"if self.debug:",
"raise AssertionError",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"if TYPE_CHECKING:",
"class .*\\bProtocol\\):"
]
```
--------------------------------------------------------------------------------
/tests/TEST_CLEANUP_PLAN.md:
--------------------------------------------------------------------------------
```markdown
# Test Cleanup Plan - Bye Weeks
## Analysis Summary
**Files Analyzed**:
- `tests/unit/test_bye_weeks.py` (637 lines, 26 tests)
- `tests/unit/test_bye_weeks_utility.py` (270 lines, 19 tests)
**Total Tests**: 45 (all passing ✅)
## Findings
### ✅ No Significant Redundancy Found
The two test files serve **distinct purposes**:
| File | Purpose | Focus | Tests |
|------|---------|-------|-------|
| `test_bye_weeks_utility.py` | Unit tests | Utility module in isolation | 19 |
| `test_bye_weeks.py` | Integration tests | Full system behavior | 26 |
### Test Coverage Breakdown
#### test_bye_weeks_utility.py (Unit Tests)
Tests the `src/utils/bye_weeks.py` module:
- ✅ `load_static_bye_weeks()` - 5 tests
- ✅ `get_bye_week_with_fallback()` - 7 tests
- ✅ `build_team_bye_week_map()` - 4 tests
- ✅ `clear_cache()` - 1 test
- ✅ Integration scenarios - 2 tests
#### test_bye_weeks.py (Integration Tests)
Tests full data flow across modules:
- ✅ `detect_bye_week()` in player_enhancement - 11 tests
- ✅ `parse_yahoo_free_agent_players()` in parsers - 11 tests
- ✅ `enhance_player_with_context()` - 2 tests
- ✅ Main functions (waiver wire, draft) - 2 tests
## Recommended Actions
### 1. Minor Documentation Cleanup ✏️
**Current State**: Verbose docstrings with repetitive explanations
**Proposed**: Streamline docstrings to be more concise
**Example**:
```python
# Before (verbose)
def test_get_waiver_wire_players_bye_week_extraction(self):
"""Test that get_waiver_wire_players correctly extracts and validates bye weeks.
Now with static fallback: invalid API data falls back to static 2025 bye weeks.
"""
# After (concise)
def test_get_waiver_wire_players_bye_week_extraction(self):
"""Test bye week extraction with static data fallback."""
```
**Impact**: Minimal - improves readability without changing functionality
### 2. Update Test Expectations Comments 📝
Some test comments still reference old behavior. Update to reflect static-first approach:
```python
# Before
# Returns: 10 (from static data, ignoring API value of 7)
# After
# Returns: 10 (static data is authoritative)
```
**Impact**: Documentation accuracy only
### 3. No Code Changes Needed ✅
**Reason**:
- Tests are well-structured and serve distinct purposes
- No actual redundancy in test logic
- All 45 tests provide valuable coverage
- Separation of unit vs integration tests is appropriate
## Conclusion
**Recommendation**: **KEEP BOTH FILES AS-IS** with only minor documentation updates.
### Why Keep Both Files?
1. **Clear Separation of Concerns**
- Unit tests focus on utility module
- Integration tests verify full system behavior
2. **Comprehensive Coverage**
- 95%+ code coverage across all modules
- Tests cover different aspects (unit vs integration)
3. **Maintainability**
- Easy to locate relevant tests
- Clear test organization by module/layer
- Well-documented test cases
4. **No Redundancy**
- Tests don't duplicate assertions
- Each test has unique purpose
- Coverage is complementary, not overlapping
## Action Items
- [x] Analyze test files for redundancy
- [x] Create cleanup plan
- [x] Document findings
- [ ] Apply minor documentation updates (optional)
- [ ] Run full test suite to verify (pytest tests/unit/ -k "bye" -v)
## Test Suite Health
| Metric | Status |
|--------|--------|
| Total Tests | 45 ✅ |
| Passing | 45 (100%) ✅ |
| Coverage | 95%+ ✅ |
| Redundancy | None found ✅ |
| Organization | Well-structured ✅ |
| Documentation | Good (minor improvements possible) ✅ |
## Final Recommendation
**Status**: ✅ **NO CLEANUP REQUIRED**
The test suite is well-organized, comprehensive, and serves its purpose effectively. The two files are complementary rather than redundant. Minor documentation improvements are optional but not necessary.
**Next Steps**:
1. Update this plan in documentation
2. Run final test verification
3. Consider complete ✅
```
--------------------------------------------------------------------------------
/src/handlers/draft_handlers.py:
--------------------------------------------------------------------------------
```python
"""Draft MCP tool handlers."""
from typing import Optional
# These will be injected from main file
get_all_teams_info = None
get_draft_rankings = None
get_draft_recommendation_simple = None
analyze_draft_state_simple = None
DRAFT_AVAILABLE = True
async def handle_ff_get_draft_results(arguments: dict) -> dict:
"""Get draft results showing all teams and their draft info.
Args:
arguments: Dict containing:
- league_key: League identifier (required)
Returns:
Dict with draft results for all teams
"""
if not arguments.get("league_key"):
return {"error": "league_key is required"}
league_key: Optional[str] = arguments.get("league_key")
if league_key is None:
return {"error": "league_key cannot be None"}
teams = await get_all_teams_info(league_key)
if not teams:
return {"error": f"Could not retrieve draft results for league {league_key}"}
return {
"league_key": league_key,
"total_teams": len(teams),
"draft_results": teams,
}
async def handle_ff_get_draft_rankings(arguments: dict) -> dict:
"""Get draft rankings with ADP data.
Args:
arguments: Dict containing:
- league_key: League identifier (required)
- position: Filter by position (default: "all")
- count: Number of players (default: 50)
Returns:
Dict with draft rankings
"""
league_key = arguments.get("league_key")
position = arguments.get("position", "all")
count = arguments.get("count", 50)
players = await get_draft_rankings(league_key, position, count)
if players:
return {
"position": position,
"total_players": len(players),
"rankings": players,
}
return {"message": "Could not retrieve draft rankings"}
async def handle_ff_get_draft_recommendation(arguments: dict) -> dict:
"""Get draft recommendations based on strategy.
Args:
arguments: Dict containing:
- league_key: League identifier (required)
- strategy: "conservative", "aggressive", or "balanced" (default: "balanced")
- num_recommendations: Number of recommendations (default: 10, max: 20)
- current_pick: Current pick number (optional)
Returns:
Dict with draft recommendations
"""
if not DRAFT_AVAILABLE:
return {"error": "Draft functionality not available. Please check module dependencies."}
try:
league_key: Optional[str] = arguments.get("league_key")
if league_key is None:
return {"error": "league_key is required and cannot be None"}
strategy = arguments.get("strategy", "balanced")
num_recommendations = arguments.get("num_recommendations", 10)
current_pick = arguments.get("current_pick")
return await get_draft_recommendation_simple(
league_key,
strategy,
num_recommendations,
current_pick,
)
except Exception as exc:
return {
"error": f"Draft recommendation failed: {exc}",
"available_tools": ["ff_get_draft_rankings", "ff_get_players"],
}
async def handle_ff_analyze_draft_state(arguments: dict) -> dict:
"""Analyze current draft state and provide strategic advice.
Args:
arguments: Dict containing:
- league_key: League identifier (required)
- strategy: "conservative", "aggressive", or "balanced" (default: "balanced")
Returns:
Dict with draft state analysis
"""
if not DRAFT_AVAILABLE:
return {"error": "Draft functionality not available. Please check module dependencies."}
try:
league_key: Optional[str] = arguments.get("league_key")
if league_key is None:
return {"error": "league_key is required and cannot be None"}
strategy = arguments.get("strategy", "balanced")
return await analyze_draft_state_simple(league_key, strategy)
except Exception as exc:
return {
"error": f"Draft analysis failed: {exc}",
"suggestion": "Try using ff_get_roster to check current team composition",
}
```
--------------------------------------------------------------------------------
/src/handlers/__init__.py:
--------------------------------------------------------------------------------
```python
"""MCP tool handlers - orchestrates handler modules with dependency injection."""
# Import simple handlers from dedicated modules (no dependencies)
from .admin_handlers import (
handle_ff_clear_cache,
handle_ff_get_api_status,
handle_ff_refresh_token,
)
# League handlers (need helper function injection)
from .league_handlers import (
handle_ff_get_league_info,
handle_ff_get_leagues,
handle_ff_get_standings,
handle_ff_get_teams,
)
# Roster handlers (need dependency injection)
from .roster_handlers import handle_ff_get_roster
# Matchup handlers (need dependency injection)
from .matchup_handlers import (
handle_ff_build_lineup,
handle_ff_compare_teams,
handle_ff_get_matchup,
)
# Player handlers (need dependency injection)
from .player_handlers import (
handle_ff_get_players,
handle_ff_get_waiver_wire,
)
# Draft handlers (need dependency injection)
from .draft_handlers import (
handle_ff_analyze_draft_state,
handle_ff_get_draft_rankings,
handle_ff_get_draft_recommendation,
handle_ff_get_draft_results,
)
# Analytics handlers (minimal dependencies)
from .analytics_handlers import handle_ff_analyze_reddit_sentiment
def inject_roster_dependencies(**deps):
"""Inject dependencies needed by roster handlers.
Required dependencies:
- get_user_team_info: Get user's team info in a league
- yahoo_api_call: Make Yahoo API calls
- parse_team_roster: Parse roster from Yahoo API response
"""
import src.handlers.roster_handlers as roster_mod
for name, func in deps.items():
setattr(roster_mod, name, func)
def inject_matchup_dependencies(**deps):
"""Inject dependencies needed by matchup handlers.
Required dependencies:
- get_user_team_key: Get user's team key in a league
- get_user_team_info: Get user's team info in a league
- yahoo_api_call: Make Yahoo API calls
- parse_team_roster: Parse roster from Yahoo API response
"""
import src.handlers.matchup_handlers as matchup_mod
for name, func in deps.items():
setattr(matchup_mod, name, func)
def inject_player_dependencies(**deps):
"""Inject dependencies needed by player handlers.
Required dependencies:
- yahoo_api_call: Make Yahoo API calls
- get_waiver_wire_players: Get waiver wire players
"""
import src.handlers.player_handlers as player_mod
for name, func in deps.items():
setattr(player_mod, name, func)
def inject_draft_dependencies(**deps):
"""Inject dependencies needed by draft handlers.
Required dependencies:
- get_all_teams_info: Get all teams info
- get_draft_rankings: Get draft rankings
- get_draft_recommendation_simple: Get draft recommendations
- analyze_draft_state_simple: Analyze draft state
- DRAFT_AVAILABLE: Draft availability flag
"""
import src.handlers.draft_handlers as draft_mod
for name, value in deps.items():
setattr(draft_mod, name, value)
def inject_league_helpers(**helpers):
"""Inject helper functions needed by league handlers.
League handlers need access to discover_leagues, get_user_team_info,
and get_all_teams_info which use global state.
"""
import src.handlers.league_handlers as league_mod
for name, func in helpers.items():
setattr(league_mod, name, func)
__all__ = [
# Admin handlers (fully extracted)
"handle_ff_refresh_token",
"handle_ff_get_api_status",
"handle_ff_clear_cache",
# League handlers (extracted, need helper injection)
"handle_ff_get_leagues",
"handle_ff_get_league_info",
"handle_ff_get_standings",
"handle_ff_get_teams",
# Roster handlers (extracted, need dependency injection)
"handle_ff_get_roster",
# Matchup handlers (extracted, need dependency injection)
"handle_ff_get_matchup",
"handle_ff_build_lineup",
"handle_ff_compare_teams",
# Player handlers (extracted, need dependency injection)
"handle_ff_get_players",
"handle_ff_get_waiver_wire",
# Draft handlers (extracted, need dependency injection)
"handle_ff_get_draft_results",
"handle_ff_get_draft_rankings",
"handle_ff_get_draft_recommendation",
"handle_ff_analyze_draft_state",
# Analytics handlers (extracted, minimal dependencies)
"handle_ff_analyze_reddit_sentiment",
# Injection functions
"inject_roster_dependencies",
"inject_matchup_dependencies",
"inject_player_dependencies",
"inject_draft_dependencies",
"inject_league_helpers",
]
```
--------------------------------------------------------------------------------
/tests/unit/test_handlers.py:
--------------------------------------------------------------------------------
```python
"""Unit tests for MCP tool handlers."""
from unittest.mock import AsyncMock, patch
import pytest
from src.handlers.admin_handlers import (
handle_ff_clear_cache,
handle_ff_get_api_status,
handle_ff_refresh_token,
)
class TestAdminHandlers:
"""Test admin and system management handlers."""
@pytest.mark.asyncio
async def test_refresh_token_success(self):
"""Test successful token refresh."""
mock_refresh = AsyncMock(
return_value={
"status": "success",
"message": "Token refreshed successfully",
"expires_in": 3600,
}
)
with patch("src.handlers.admin_handlers.refresh_yahoo_token", mock_refresh):
result = await handle_ff_refresh_token({})
assert result["status"] == "success"
assert "expires_in" in result
mock_refresh.assert_called_once()
@pytest.mark.asyncio
async def test_refresh_token_failure(self):
"""Test failed token refresh."""
mock_refresh = AsyncMock(return_value={"status": "error", "message": "Missing credentials"})
with patch("src.handlers.admin_handlers.refresh_yahoo_token", mock_refresh):
result = await handle_ff_refresh_token({})
assert result["status"] == "error"
assert "Missing credentials" in result["message"]
@pytest.mark.asyncio
async def test_get_api_status(self, mock_rate_limiter, mock_response_cache):
"""Test getting API status."""
with (
patch("src.handlers.admin_handlers.rate_limiter", mock_rate_limiter),
patch("src.handlers.admin_handlers.response_cache", mock_response_cache),
):
result = await handle_ff_get_api_status({})
assert "rate_limit" in result
assert "cache" in result
assert result["rate_limit"]["requests_remaining"] == 850
assert result["cache"]["hits"] == 245
@pytest.mark.asyncio
async def test_clear_cache_full(self, mock_response_cache):
"""Test clearing entire cache."""
with patch("src.handlers.admin_handlers.response_cache", mock_response_cache):
result = await handle_ff_clear_cache({})
assert result["status"] == "success"
assert "completely" in result["message"]
mock_response_cache.clear.assert_called_once_with(None)
@pytest.mark.asyncio
async def test_clear_cache_with_pattern(self, mock_response_cache):
"""Test clearing cache with pattern."""
with patch("src.handlers.admin_handlers.response_cache", mock_response_cache):
result = await handle_ff_clear_cache({"pattern": "league/*"})
assert result["status"] == "success"
assert "league/*" in result["message"]
mock_response_cache.clear.assert_called_once_with("league/*")
class TestLeagueHandlers:
"""Test league-level handlers."""
@pytest.mark.asyncio
async def test_get_leagues_error_handling(self):
"""Test error handling when league_key is missing."""
from src.handlers.league_handlers import handle_ff_get_league_info
result = await handle_ff_get_league_info({})
assert "error" in result
assert "league_key is required" in result["error"]
@pytest.mark.asyncio
async def test_get_standings_error_handling(self):
"""Test error handling for standings without league_key."""
from src.handlers.league_handlers import handle_ff_get_standings
result = await handle_ff_get_standings({})
assert "error" in result
assert "league_key is required" in result["error"]
@pytest.mark.asyncio
async def test_get_teams_error_handling(self):
"""Test error handling for teams without league_key."""
from src.handlers.league_handlers import handle_ff_get_teams
result = await handle_ff_get_teams({})
assert "error" in result
assert "league_key is required" in result["error"]
@pytest.mark.asyncio
async def test_get_standings_parses_correctly(self, mock_yahoo_standings_response):
"""Test standings parsing logic."""
from src.handlers.league_handlers import handle_ff_get_standings
mock_api_call = AsyncMock(return_value=mock_yahoo_standings_response)
with patch("src.handlers.league_handlers.yahoo_api_call", mock_api_call):
result = await handle_ff_get_standings({"league_key": "461.l.61410"})
assert "standings" in result
assert len(result["standings"]) == 2
assert result["standings"][0]["team"] == "Team Alpha"
assert result["standings"][0]["rank"] == 1
assert result["standings"][0]["wins"] == 10
assert result["standings"][1]["team"] == "Team Bravo"
assert result["standings"][1]["rank"] == 2
```
--------------------------------------------------------------------------------
/src/utils/bye_weeks.py:
--------------------------------------------------------------------------------
```python
"""
Utility module for loading and managing NFL bye week data.
Provides static bye week data as a fallback when API data is missing or invalid.
"""
import json
import logging
from pathlib import Path
from typing import Dict, Optional
logger = logging.getLogger(__name__)
# Cache for loaded bye week data to avoid repeated file reads
_BYE_WEEK_CACHE: Optional[Dict[str, int]] = None
def load_static_bye_weeks() -> Dict[str, int]:
"""
Load static bye week data from JSON file.
Returns:
Dictionary mapping team abbreviations to bye week numbers.
Returns empty dict if file cannot be loaded.
"""
global _BYE_WEEK_CACHE
# Return cached data if available
if _BYE_WEEK_CACHE is not None:
return _BYE_WEEK_CACHE
try:
# Get the path to the static data file
data_file = Path(__file__).parent.parent / "data" / "bye_weeks_2025.json"
with open(data_file, 'r') as f:
bye_weeks = json.load(f)
# Validate the data structure
if not isinstance(bye_weeks, dict):
logger.error("Static bye week data is not a dictionary")
return {}
# Validate all values are integers between 1 and 18
for team, week in bye_weeks.items():
if not isinstance(week, int) or not (1 <= week <= 18):
logger.warning(f"Invalid bye week {week} for team {team} in static data")
# Cache the loaded data
_BYE_WEEK_CACHE = bye_weeks
logger.info(f"Loaded static bye week data for {len(bye_weeks)} teams")
return bye_weeks
except FileNotFoundError:
logger.error("Static bye week data file not found")
return {}
except json.JSONDecodeError as e:
logger.error(f"Error parsing static bye week data: {e}")
return {}
except Exception as e:
logger.error(f"Unexpected error loading static bye week data: {e}")
return {}
def get_bye_week_with_fallback(
team_abbr: str,
api_bye_week: Optional[int] = None
) -> Optional[int]:
"""
Get bye week for a team, preferring static data as authoritative source.
Static data contains the correct 2025 NFL bye weeks and is always used when available.
API data is only used as a fallback if the team is not in static data.
Args:
team_abbr: Team abbreviation (e.g., "KC", "SF", "BUF")
api_bye_week: Bye week from API (if available, used only as fallback)
Returns:
Bye week number (1-18) or None if not found.
"""
# Load static data (authoritative source for 2025)
static_data = load_static_bye_weeks()
# Always prefer static data when available
if team_abbr in static_data:
bye_week = static_data[team_abbr]
if api_bye_week is not None and api_bye_week != bye_week and 1 <= api_bye_week <= 18:
logger.debug(
f"Using static bye week {bye_week} for {team_abbr} "
f"(overriding API value: {api_bye_week})"
)
return bye_week
# Fall back to API data only if team not in static data
if api_bye_week is not None and isinstance(api_bye_week, int) and 1 <= api_bye_week <= 18:
logger.info(
f"Using API bye week {api_bye_week} for {team_abbr} "
f"(team not in static data)"
)
return api_bye_week
logger.warning(f"No bye week data found for team {team_abbr} (static or API)")
return None
def build_team_bye_week_map(
api_team_data: Optional[Dict[str, int]] = None
) -> Dict[str, int]:
"""
Build a complete team-to-bye-week mapping with fallback support.
Combines API data (if available) with static data to ensure all teams
have bye week information.
Args:
api_team_data: Optional dictionary of team abbreviations to bye weeks from API
Returns:
Dictionary mapping team abbreviations to bye week numbers.
"""
# Start with static data as baseline
bye_week_map = load_static_bye_weeks().copy()
# Override with API data where available and valid
if api_team_data:
valid_count = 0
invalid_count = 0
for team, week in api_team_data.items():
if isinstance(week, int) and 1 <= week <= 18:
bye_week_map[team] = week
valid_count += 1
else:
invalid_count += 1
logger.warning(f"Ignoring invalid API bye week {week} for {team}")
if valid_count > 0:
logger.info(f"Updated {valid_count} teams with API bye week data")
if invalid_count > 0:
logger.info(f"Kept static data for {invalid_count} teams due to invalid API data")
else:
logger.info("No API bye week data provided, using all static data")
return bye_week_map
def clear_cache():
"""Clear the cached bye week data. Useful for testing or forcing a reload."""
global _BYE_WEEK_CACHE
_BYE_WEEK_CACHE = None
logger.debug("Bye week cache cleared")
```
--------------------------------------------------------------------------------
/src/api/yahoo_client.py:
--------------------------------------------------------------------------------
```python
"""Yahoo Fantasy Sports API client with rate limiting and token refresh."""
import os
import socket
from typing import Dict
import aiohttp
from src.api.yahoo_utils import rate_limiter, response_cache
# Module-level token cache
_YAHOO_ACCESS_TOKEN = os.getenv("YAHOO_ACCESS_TOKEN")
YAHOO_API_BASE = "https://fantasysports.yahooapis.com/fantasy/v2"
def get_access_token() -> str:
"""Get the current access token."""
global _YAHOO_ACCESS_TOKEN
if _YAHOO_ACCESS_TOKEN is None:
_YAHOO_ACCESS_TOKEN = os.getenv("YAHOO_ACCESS_TOKEN")
return _YAHOO_ACCESS_TOKEN or ""
def set_access_token(token: str) -> None:
"""Update the access token (used after refresh)."""
global _YAHOO_ACCESS_TOKEN
_YAHOO_ACCESS_TOKEN = token
os.environ["YAHOO_ACCESS_TOKEN"] = token
async def yahoo_api_call(
endpoint: str, retry_on_auth_fail: bool = True, use_cache: bool = True
) -> Dict:
"""Make Yahoo API request with rate limiting, caching, and automatic token refresh.
Args:
endpoint: Yahoo API endpoint (e.g., "users;use_login=1/games")
retry_on_auth_fail: If True, will attempt token refresh on 401 errors
use_cache: If True, will check cache before making API call
Returns:
dict: JSON response from Yahoo API
Raises:
Exception: On API errors or authentication failures
"""
# Check cache first (if enabled)
if use_cache:
cached_response = await response_cache.get(endpoint)
if cached_response is not None:
return cached_response
# Apply rate limiting
await rate_limiter.acquire()
access_token = get_access_token()
url = f"{YAHOO_API_BASE}/{endpoint}?format=json"
headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"}
connector = aiohttp.TCPConnector(family=socket.AF_INET)
async with aiohttp.ClientSession(connector=connector, trust_env=True) as session:
async with session.get(url, headers=headers) as response:
if response.status == 200:
data = await response.json()
# Cache successful response
if use_cache:
await response_cache.set(endpoint, data)
return data
elif response.status == 401 and retry_on_auth_fail:
# Token expired, try to refresh
refresh_result = await refresh_yahoo_token()
if refresh_result.get("status") == "success":
# Token refreshed, retry the API call with new token
return await yahoo_api_call(
endpoint, retry_on_auth_fail=False, use_cache=use_cache
)
else:
# Refresh failed, raise the original error
text = await response.text()
raise Exception(f"Yahoo API auth failed and token refresh failed: {text[:200]}")
else:
text = await response.text()
raise Exception(f"Yahoo API error {response.status}: {text[:200]}")
async def refresh_yahoo_token() -> Dict:
"""Refresh the Yahoo access token using the refresh token.
Returns:
dict: Status message with refresh result
- {"status": "success", "message": "...", "expires_in": 3600}
- {"status": "error", "message": "...", "details": "..."}
"""
client_id = os.getenv("YAHOO_CONSUMER_KEY")
client_secret = os.getenv("YAHOO_CONSUMER_SECRET")
refresh_token = os.getenv("YAHOO_REFRESH_TOKEN")
if not all([client_id, client_secret, refresh_token]):
return {"status": "error", "message": "Missing credentials in environment"}
token_url = "https://api.login.yahoo.com/oauth2/get_token"
data = {
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token",
}
try:
connector = aiohttp.TCPConnector(family=socket.AF_INET)
async with aiohttp.ClientSession(connector=connector, trust_env=True) as session:
async with session.post(token_url, data=data) as response:
if response.status == 200:
token_data = await response.json()
new_access_token = token_data.get("access_token")
new_refresh_token = token_data.get("refresh_token", refresh_token)
expires_in = token_data.get("expires_in", 3600)
# Update global token
set_access_token(new_access_token)
# Update environment
if new_refresh_token != refresh_token:
os.environ["YAHOO_REFRESH_TOKEN"] = new_refresh_token
return {
"status": "success",
"message": "Token refreshed successfully",
"expires_in": expires_in,
"expires_in_hours": round(expires_in / 3600, 1),
}
else:
error_text = await response.text()
return {
"status": "error",
"message": f"Failed to refresh token: {response.status}",
"details": error_text[:200],
}
except Exception as e:
return {"status": "error", "message": f"Error refreshing token: {str(e)}"}
```
--------------------------------------------------------------------------------
/utils/setup_yahoo_auth.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
One-time Yahoo Fantasy API Authentication Setup
Run this script once to authenticate and save your token.
"""
import os
import sys
import json
import webbrowser
from pathlib import Path
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
print("=" * 70)
print("🏈 YAHOO FANTASY API - ONE-TIME AUTHENTICATION SETUP")
print("=" * 70)
print()
# Your credentials from .env
CLIENT_ID = os.getenv("YAHOO_CLIENT_ID")
CLIENT_SECRET = os.getenv("YAHOO_CLIENT_SECRET")
if not CLIENT_ID or not CLIENT_SECRET:
print("❌ ERROR: Yahoo credentials not found in .env file")
print("Please make sure your .env file contains:")
print(" YAHOO_CLIENT_ID=your_client_id")
print(" YAHOO_CLIENT_SECRET=your_client_secret")
sys.exit(1)
print("✅ Found Yahoo credentials")
print(f" Client ID: {CLIENT_ID[:30]}...")
print(f" Client Secret: {CLIENT_SECRET[:10]}...")
print()
# Method 1: Using yfpy (Recommended)
print("METHOD 1: Using yfpy Library (Recommended)")
print("-" * 40)
try:
from yfpy import YahooFantasySportsQuery
print("This will:")
print("1. Open your browser to Yahoo login")
print("2. You login and click 'Agree' to authorize")
print("3. Yahoo will show a verification code")
print("4. Come back here and paste that code")
print()
input("Press Enter to start the authentication process...")
print()
# Create token directory
token_dir = Path(".tokens")
token_dir.mkdir(exist_ok=True)
print("🌐 Opening browser for Yahoo authorization...")
print()
# Initialize - this will trigger OAuth flow
try:
query = YahooFantasySportsQuery(
league_id="", # Empty to get all leagues
game_code="nfl",
game_id=449, # 2025 NFL season
yahoo_consumer_key=CLIENT_ID,
yahoo_consumer_secret=CLIENT_SECRET,
browser_callback=True, # Opens browser automatically
env_file_location=Path("."), # Save token to current directory
save_token_data_to_env_file=True # Save for reuse
)
print()
print("✅ Authentication successful!")
print()
# Test by getting user leagues
print("Testing connection by fetching your leagues...")
try:
# Get user info to verify connection
user_games = query.get_user_games()
print(f"✅ Connected! Found {len(user_games) if user_games else 0} games")
# Try to get leagues
user_leagues = query.get_user_leagues_by_game_key("449") # NFL 2025 season
if user_leagues:
print(f"✅ Found {len(user_leagues)} leagues:")
for i, league in enumerate(user_leagues, 1):
league_name = getattr(league, 'name', 'Unknown')
league_id = getattr(league, 'league_id', 'Unknown')
print(f" {i}. {league_name} (ID: {league_id})")
# Save token for MCP server use
token_file = Path(".yahoo_token.json")
if hasattr(query, 'oauth') and hasattr(query.oauth, 'token_data'):
with open(token_file, 'w') as f:
json.dump(query.oauth.token_data, f, indent=2)
print(f"\n✅ Token saved to {token_file}")
print(" The MCP server can now use this token!")
except Exception as e:
print(f"⚠️ Connection test failed: {e}")
print(" But authentication may still be successful.")
except Exception as e:
print(f"\n❌ Authentication failed: {e}")
print()
print("Troubleshooting:")
print("1. Make sure your Yahoo app is configured correctly:")
print(" - Go to https://developer.yahoo.com/apps/")
print(" - Check your app has 'Fantasy Sports - Read' permission")
print(" - Redirect URI should be: oob (for out-of-band)")
print("2. Try deleting any .yahoo_oauth or token files and retry")
except ImportError:
print("❌ yfpy not installed")
print("Install with: pip install yfpy")
print()
print("Falling back to Method 2...")
print()
# Method 2: Manual OAuth flow
print("METHOD 2: Manual OAuth Flow")
print("-" * 40)
print()
# Build authorization URL
auth_url = (
"https://api.login.yahoo.com/oauth2/request_auth?"
f"client_id={CLIENT_ID}&"
"redirect_uri=oob&"
"response_type=code&"
"language=en-us"
)
print("Manual authentication steps:")
print()
print("1. Copy this URL and open it in your browser:")
print()
print(auth_url)
print()
print("2. Login to Yahoo and click 'Agree'")
print("3. Yahoo will show you a verification code")
print("4. Copy that code and save it")
print()
print("5. You'll need to exchange this code for tokens")
print(" (This requires additional implementation)")
# Try to open browser automatically
try:
webbrowser.open(auth_url)
print("✅ Browser opened automatically")
except:
print("⚠️ Could not open browser automatically")
print(" Please copy the URL above and open it manually")
print()
print("=" * 70)
print("NEXT STEPS")
print("=" * 70)
print()
print("Once authenticated:")
print("1. The token is saved to .yahoo_token.json")
print("2. The MCP server will use this token automatically")
print("3. Token will auto-refresh as needed")
print()
print("To use with MCP:")
print("1. Add to your MCP config (Claude, etc.)")
print("2. The server will use the saved token")
print("3. Start making Fantasy Football API calls!")
print()
print("Need help? Check YAHOO_AUTH_REALITY.md for more details.")
```
--------------------------------------------------------------------------------
/utils/refresh_yahoo_token.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Refresh Yahoo Fantasy Sports OAuth2 Token
"""
import os
import json
import requests
from dotenv import load_dotenv
from datetime import datetime
# Load environment
load_dotenv()
def refresh_yahoo_token():
"""Refresh the Yahoo access token using the refresh token."""
# Get credentials from environment
client_id = os.getenv("YAHOO_CONSUMER_KEY")
client_secret = os.getenv("YAHOO_CONSUMER_SECRET")
refresh_token = os.getenv("YAHOO_REFRESH_TOKEN")
if not all([client_id, client_secret, refresh_token]):
print("❌ Missing credentials in .env file")
print("Required: YAHOO_CONSUMER_KEY, YAHOO_CONSUMER_SECRET, YAHOO_REFRESH_TOKEN")
return False
# Yahoo token endpoint
token_url = "https://api.login.yahoo.com/oauth2/get_token"
# Prepare refresh request
data = {
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token",
}
print("🔄 Refreshing Yahoo token...")
try:
# Make refresh request
response = requests.post(token_url, data=data)
if response.status_code == 200:
# Parse new tokens
token_data = response.json()
new_access_token = token_data.get("access_token")
new_refresh_token = token_data.get("refresh_token", refresh_token)
expires_in = token_data.get("expires_in", 3600)
print("✅ Token refreshed successfully!")
print(f" Expires in: {expires_in} seconds ({expires_in/3600:.1f} hours)")
# Update .env file
update_env_file(new_access_token, new_refresh_token)
# Also update claude_desktop_config.json if it exists
update_claude_config(new_access_token, new_refresh_token)
print("\n📝 Updated tokens in:")
print(" - .env file")
print(" - claude_desktop_config.json")
print("\n⚠️ IMPORTANT: Restart Claude Desktop to use the new token")
return True
else:
print(f"❌ Failed to refresh token: {response.status_code}")
print(f" Response: {response.text}")
if response.status_code == 400:
print("\n💡 If refresh token is expired, you need to re-authenticate:")
print(" Run: python setup_yahoo_auth.py")
return False
except Exception as e:
print(f"❌ Error refreshing token: {e}")
return False
def update_env_file(access_token, refresh_token):
"""Update the .env file with new tokens."""
# Read current .env
env_path = ".env"
lines = []
if os.path.exists(env_path):
with open(env_path, "r") as f:
lines = f.readlines()
# Update or add token lines
updated = False
new_lines = []
for line in lines:
if line.startswith("YAHOO_ACCESS_TOKEN="):
new_lines.append(f"YAHOO_ACCESS_TOKEN={access_token}\n")
updated = True
elif line.startswith("YAHOO_REFRESH_TOKEN="):
new_lines.append(f"YAHOO_REFRESH_TOKEN={refresh_token}\n")
else:
new_lines.append(line)
# Add tokens if not found
if not updated:
new_lines.append(f"\nYAHOO_ACCESS_TOKEN={access_token}\n")
new_lines.append(f"YAHOO_REFRESH_TOKEN={refresh_token}\n")
# Write back
with open(env_path, "w") as f:
f.writelines(new_lines)
def update_claude_config(access_token, refresh_token):
"""Update the Claude Desktop config with new tokens."""
config_path = "claude_desktop_config.json"
if not os.path.exists(config_path):
return
try:
with open(config_path, "r") as f:
config = json.load(f)
# Update tokens in the fantasy-football server env
if "mcpServers" in config and "fantasy-football" in config["mcpServers"]:
if "env" not in config["mcpServers"]["fantasy-football"]:
config["mcpServers"]["fantasy-football"]["env"] = {}
config["mcpServers"]["fantasy-football"]["env"]["YAHOO_ACCESS_TOKEN"] = access_token
config["mcpServers"]["fantasy-football"]["env"]["YAHOO_REFRESH_TOKEN"] = refresh_token
# Write back
with open(config_path, "w") as f:
json.dump(config, f, indent=4)
except Exception as e:
print(f"⚠️ Could not update Claude config: {e}")
def test_new_token():
"""Test if the new token works."""
load_dotenv(override=True) # Reload environment
access_token = os.getenv("YAHOO_ACCESS_TOKEN")
if not access_token:
print("❌ No access token found")
return False
# Test API call
url = "https://fantasysports.yahooapis.com/fantasy/v2/users;use_login=1?format=json"
headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"}
try:
response = requests.get(url, headers=headers)
if response.status_code == 200:
print("\n✅ Token test successful! API is accessible.")
return True
else:
print(f"\n❌ Token test failed: {response.status_code}")
return False
except Exception as e:
print(f"\n❌ Token test error: {e}")
return False
if __name__ == "__main__":
print("=" * 60)
print("Yahoo Fantasy Sports Token Refresh")
print("=" * 60)
print()
if refresh_yahoo_token():
print("\n🧪 Testing new token...")
test_new_token()
print("\n" + "=" * 60)
print("✅ Token refresh complete!")
print("=" * 60)
else:
print("\n" + "=" * 60)
print("❌ Token refresh failed")
print("=" * 60)
print("\nTroubleshooting:")
print("1. Check your internet connection")
print("2. Verify your credentials in .env")
print("3. If refresh token is expired, run: python setup_yahoo_auth.py")
print("4. Check Yahoo Developer App settings")
```
--------------------------------------------------------------------------------
/docs/PROMPTS_AND_RESOURCES.md:
--------------------------------------------------------------------------------
```markdown
# MCP Prompts & Resources - Feature Summary
## Overview
Added comprehensive MCP prompts and resources to enable LLMs to provide better fantasy football advice through structured templates and contextual knowledge.
## What Was Added
### ✅ 8 New Prompt Templates
1. **`start_sit_decision`** - Start/sit decision making with confidence levels
- Parameters: `league_key`, `position`, `player_names`, `week`
- Analyzes: projections, matchups, trends, injuries, weather, usage, game script
2. **`bye_week_planning`** - Bye week planning and roster management
- Parameters: `league_key`, `team_key`, `upcoming_weeks`
- Provides: week-by-week action plan for bye week coverage
3. **`playoff_preparation`** - Championship preparation strategy
- Parameters: `league_key`, `team_key`, `current_week`
- Focuses on: playoff schedules, key acquisitions, handcuffs, weather considerations
4. **`trade_proposal_generation`** - Fair trade proposal creation
- Parameters: `league_key`, `my_team_key`, `target_team_key`, `position_need`
- Creates: 2-3 trade options with reasoning for both teams
5. **`injury_replacement_strategy`** - Comprehensive injury replacement plans
- Parameters: `league_key`, `injured_player`, `injury_length`, `position`
- Provides: short/long-term strategies, waiver targets, FAAB bids, timeline
6. **`streaming_dst_kicker`** - Weekly streaming recommendations
- Parameters: `league_key`, `week`, `position` (DEF/K)
- Analyzes: matchups, weather, Vegas lines, 2-3 week preview
7. **`season_long_strategy_check`** - Overall season assessment
- Parameters: `league_key`, `team_key`, `current_record`, `weeks_remaining`
- Evaluates: playoff probability, trade deadline strategy, must-win games
8. **`weekly_game_plan`** - Complete weekly game plan
- Parameters: `league_key`, `team_key`, `opponent_team_key`, `week`
- Creates: full action plan with lineup, start/sit, opponent analysis, waiver claims
### ✅ 5 New Resource Guides
1. **`guide://weekly-strategy`** - Week-by-week strategic guidance
- Covers: Weeks 1-4 (early season) through Weeks 15-17 (playoffs)
- Includes: weekly task checklist for each phase of season
2. **`guide://common-mistakes`** - Common fantasy football mistakes
- Categories: Draft, In-season, Waiver wire, Trade, Lineup, Strategic
- Provides: ❌ Mistakes to avoid and ✅ Best practices
3. **`guide://advanced-stats`** - Advanced statistics glossary
- Covers: Volume metrics, efficiency metrics, situation metrics, opportunity metrics
- Examples: Snap %, target share, YPRR, air yards, game script, red zone touches
4. **`guide://playoff-strategies`** - Championship preparation tactics
- Covers: Roster construction, schedule analysis, positional strategy
- Includes: Week 17 rest considerations, weather impact, streaming strategies
5. **`guide://dynasty-keeper`** - Dynasty and keeper league strategies
- Covers: Valuation differences, rookie drafts, aging curves, trade windows
- Includes: Contender vs rebuilder strategies, keeper value calculations
### ✅ Existing Features (Already Present)
**Prompts:**
- `analyze_roster_strengths`
- `draft_strategy_advice`
- `matchup_analysis`
- `waiver_wire_priority`
- `trade_evaluation`
**Resources:**
- `config://scoring` - Standard/PPR scoring rules
- `config://positions` - Position requirements
- `config://strategies` - Draft strategies
- `data://injury-status` - Injury designation guide
- `guide://tool-selection` - Tool usage guide for LLMs
- `meta://version` - Server version info
## Total Count
- **13 Prompt Templates** (5 existing + 8 new)
- **11 Resource Guides** (6 existing + 5 new)
## Updated Files
1. **`fastmcp_server.py`**
- Added 8 new `@server.prompt` decorated functions
- Added 5 new `@server.resource` decorated functions
- Updated `__all__` export list with all new prompts and resources
- Total additions: ~580 lines
2. **`CLAUDE.md`**
- Added comprehensive "MCP Prompts & Resources" section
- Documents all available prompts and resources
- Includes usage examples and benefits for LLMs
- Total additions: ~90 lines
## How LLMs Use These
### Prompts
Prompts provide structured templates that guide LLMs to:
- Ask the right questions
- Consider all relevant factors
- Provide consistent output formats
- Include confidence levels and reasoning
Example usage flow:
```
User: "Should I start Derrick Henry or Najee Harris?"
↓
LLM uses start_sit_decision prompt
↓
LLM asks for: projections, matchups, trends, injuries, weather, etc.
↓
LLM provides: structured recommendation with confidence level
```
### Resources
Resources provide domain knowledge that LLMs can reference:
- Scoring rules (understand point values)
- Position requirements (roster construction)
- Strategy frameworks (best practices)
- Common mistakes (pitfalls to avoid)
- Advanced metrics (stat interpretations)
Example usage flow:
```
User: "What's a good target share for a WR1?"
↓
LLM accesses guide://advanced-stats resource
↓
LLM knows: "20%+ is WR1 territory, 25%+ is elite"
↓
LLM provides: informed answer with context
```
## Benefits
1. **Consistency** - All LLMs get the same structured approach
2. **Completeness** - Prompts ensure no factors are overlooked
3. **Context** - Resources provide domain expertise without training
4. **Quality** - Better recommendations through better frameworks
5. **Scalability** - Easy to add new prompts/resources as needed
## Testing
✅ Python syntax validation passed
✅ 13 prompt templates detected and validated
✅ 11 resource guides present and accessible
✅ Updated `__all__` export list
✅ Documentation updated in CLAUDE.md
## Next Steps (Optional Future Enhancements)
- Add more sport-specific prompts (NFL team analysis, schedule strength)
- Create position-specific resource guides (RB strategies, WR route concepts)
- Add dynasty-specific prompts (rookie draft recommendations, trade calculator)
- Create league format resources (Best Ball, DFS, Superflex strategies)
- Add historical data resources (ADP trends, breakout patterns)
```
--------------------------------------------------------------------------------
/src/handlers/league_handlers.py:
--------------------------------------------------------------------------------
```python
"""League-level MCP tool handlers (leagues, standings, teams)."""
from typing import Dict, Optional
from src.api import yahoo_api_call
# These functions need to be imported from main file since they use global cache
# We'll import them when updating fantasy_football_multi_league.py
async def discover_leagues():
"""Placeholder - will be imported from main module."""
raise NotImplementedError("Must import from fantasy_football_multi_league")
async def get_user_team_info(league_key):
"""Placeholder - will be imported from main module."""
raise NotImplementedError("Must import from fantasy_football_multi_league")
async def get_all_teams_info(league_key):
"""Placeholder - will be imported from main module."""
raise NotImplementedError("Must import from fantasy_football_multi_league")
async def handle_ff_get_leagues(arguments: Dict) -> Dict:
"""Get all fantasy football leagues for the authenticated user.
Args:
arguments: Empty dict (no arguments required)
Returns:
Dict with total_leagues and list of league summaries
"""
leagues = await discover_leagues()
if not leagues:
return {
"error": "No active NFL leagues found",
"suggestion": "Make sure your Yahoo token is valid and you have active leagues",
}
return {
"total_leagues": len(leagues),
"leagues": [
{
"key": league["key"],
"name": league["name"],
"teams": league["num_teams"],
"current_week": league["current_week"],
"scoring": league["scoring_type"],
}
for league in leagues.values()
],
}
async def handle_ff_get_league_info(arguments: Dict) -> Dict:
"""Get detailed information about a specific league.
Args:
arguments: Dict with 'league_key'
Returns:
Dict with league details and your team summary
"""
if not arguments.get("league_key"):
return {"error": "league_key is required"}
league_key = arguments.get("league_key")
leagues = await discover_leagues()
if league_key not in leagues:
return {
"error": f"League {league_key} not found",
"available_leagues": list(leagues.keys()),
}
league = leagues[league_key]
team_info = await get_user_team_info(league_key)
_ = await yahoo_api_call(f"league/{league_key}")
return {
"league": league["name"],
"key": league_key,
"season": league["season"],
"teams": league["num_teams"],
"current_week": league["current_week"],
"scoring_type": league["scoring_type"],
"status": "active" if not league["is_finished"] else "finished",
"your_team": {
"name": team_info.get("team_name", "Unknown") if team_info else "Not found",
"key": team_info.get("team_key") if team_info else None,
"draft_position": team_info.get("draft_position") if team_info else None,
"draft_grade": team_info.get("draft_grade") if team_info else None,
},
}
async def handle_ff_get_standings(arguments: Dict) -> Dict:
"""Get current standings for a league.
Args:
arguments: Dict with 'league_key'
Returns:
Dict with league_key and sorted standings list
"""
if not arguments.get("league_key"):
return {"error": "league_key is required"}
league_key = arguments.get("league_key")
data = await yahoo_api_call(f"league/{league_key}/standings")
standings = []
league = data.get("fantasy_content", {}).get("league", [])
for item in league:
if isinstance(item, dict) and "standings" in item:
standings_list = item["standings"]
teams = {}
if isinstance(standings_list, list) and standings_list:
teams = standings_list[0].get("teams", {})
elif isinstance(standings_list, dict):
teams = standings_list.get("teams", {})
for key, team_entry in teams.items():
if key == "count" or not isinstance(team_entry, dict):
continue
if "team" in team_entry:
team_array = team_entry["team"]
team_info = {}
team_standings = {}
if isinstance(team_array, list) and team_array:
core = team_array[0]
if isinstance(core, list):
for elem in core:
if isinstance(elem, dict) and "name" in elem:
team_info["name"] = elem["name"]
for part in team_array[1:]:
if isinstance(part, dict) and "team_standings" in part:
team_standings = part["team_standings"]
if team_info and team_standings:
standings.append(
{
"rank": team_standings.get("rank", 0),
"team": team_info.get("name", "Unknown"),
"wins": team_standings.get("outcome_totals", {}).get("wins", 0),
"losses": team_standings.get("outcome_totals", {}).get("losses", 0),
"ties": team_standings.get("outcome_totals", {}).get("ties", 0),
"points_for": team_standings.get("points_for", 0),
"points_against": team_standings.get("points_against", 0),
}
)
standings.sort(key=lambda row: row["rank"])
return {"league_key": league_key, "standings": standings}
async def handle_ff_get_teams(arguments: Dict) -> Dict:
"""Get all teams in a league.
Args:
arguments: Dict with 'league_key'
Returns:
Dict with league_key, teams list, and total_teams count
"""
if not arguments.get("league_key"):
return {"error": "league_key is required"}
league_key: Optional[str] = arguments.get("league_key")
if league_key is None:
return {"error": "league_key cannot be None"}
teams_info = await get_all_teams_info(league_key)
return {
"league_key": league_key,
"teams": teams_info,
"total_teams": len(teams_info),
}
```
--------------------------------------------------------------------------------
/INSTALLATION.md:
--------------------------------------------------------------------------------
```markdown
# Fantasy Football MCP Server - Installation Guide
## Prerequisites
- Python 3.8 or higher
- Claude Desktop application
- Yahoo Fantasy Sports account with active leagues
- Git (for cloning the repository)
## Step 1: Clone the Repository
```bash
git clone https://github.com/derekrbreese/fantasy-football-mcp-public.git
cd fantasy-football-mcp-public
```
## Step 2: Install Python Dependencies
```bash
pip install -r requirements.txt
```
Or if you prefer using a virtual environment:
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txt
```
## Step 3: Yahoo API Setup
### 3.1 Create a Yahoo Developer App
1. Go to https://developer.yahoo.com/apps/
2. Click "Create an App"
3. Fill in the application details:
- **Application Name**: Fantasy Football MCP (or your choice)
- **Application Type**: Web Application
- **Redirect URI(s)**: `http://localhost:8000/callback`
- **API Permissions**: Fantasy Sports (Read)
4. Click "Create App"
5. Save your **Client ID (Consumer Key)** and **Client Secret (Consumer Secret)**
### 3.2 Initial Authentication
Run the authentication script to get your tokens:
```bash
python reauth_yahoo.py
```
This will:
1. Open your browser for Yahoo login
2. Ask you to authorize the app
3. Automatically save your tokens to `.env` file
4. Display your team information to confirm it's working
## Step 4: Environment Configuration
The `.env` file should be automatically created after authentication. Verify it contains:
```env
# Yahoo API Credentials
YAHOO_CONSUMER_KEY=your_consumer_key_here
YAHOO_CONSUMER_SECRET=your_consumer_secret_here
YAHOO_ACCESS_TOKEN=your_access_token_here
YAHOO_REFRESH_TOKEN=your_refresh_token_here
YAHOO_GUID=your_yahoo_guid_here
```
**Note**: Since this is a private repository, the `.env` file is tracked for backup purposes.
## Step 5: Claude Desktop Configuration
### 5.1 Locate Claude Desktop Config
The configuration file location depends on your operating system:
- **macOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
- **Linux**: `~/.config/Claude/claude_desktop_config.json`
### 5.2 Add MCP Server Configuration
Add the following to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"fantasy-football": {
"command": "python",
"args": [
"/absolute/path/to/fantasy_football_multi_league.py"
],
"env": {
"YAHOO_ACCESS_TOKEN": "your_access_token",
"YAHOO_CONSUMER_KEY": "your_consumer_key",
"YAHOO_CONSUMER_SECRET": "your_consumer_secret",
"YAHOO_REFRESH_TOKEN": "your_refresh_token",
"YAHOO_GUID": "your_yahoo_guid"
}
}
}
}
```
**Important**:
- Replace `/absolute/path/to/` with the actual path to your installation
- Copy the credentials from your `.env` file
- If you have other MCP servers configured, add this as an additional entry
### 5.3 Alternative: Use the Provided Config
You can also copy the provided config template:
```bash
cp claude_desktop_config.json ~/Library/Application\ Support/Claude/claude_desktop_config.json
```
Then edit it to update the file paths and credentials.
## Step 6: Test the Installation
### 6.1 Test Python Server Directly
```bash
python test_multi_league_server.py
```
Expected output:
- Should find all your active leagues
- Should identify your team in each league
### 6.2 Restart Claude Desktop
After updating the configuration:
1. Completely quit Claude Desktop
2. Restart Claude Desktop
3. The MCP tools should now be available
### 6.3 Verify in Claude Desktop
Ask Claude: "Use the fantasy football tools to show me my leagues"
Claude should be able to use the `ff_get_leagues` tool and show your active leagues.
## Step 7: Token Management
### Automatic Token Refresh
The server includes automatic token refresh capability. You can also manually refresh:
**Through Claude Desktop**:
- Ask Claude to "refresh my Yahoo token"
**Through Command Line**:
```bash
python refresh_yahoo_token.py
```
### Full Re-authentication
If tokens are completely expired (after ~60 days):
```bash
python reauth_yahoo.py
```
## Available MCP Tools
Once installed, you'll have access to 12 tools:
1. **ff_get_leagues** - List all your fantasy football leagues
2. **ff_get_league_info** - Get detailed league information with your team name
3. **ff_get_standings** - View current standings
4. **ff_get_roster** - Get a team roster with team name (accepts optional `team_key`)
5. **ff_compare_teams** - Compare two teams' rosters
6. **ff_get_matchup** - View matchup details
7. **ff_get_players** - Browse available players
8. **ff_build_lineup** - Build optimal lineup with positional constraints
9. **ff_refresh_token** - Refresh Yahoo access token
10. **ff_get_draft_results** - View draft results and grades
11. **ff_get_waiver_wire** - Find top waiver wire pickups
12. **ff_get_draft_rankings** - Get pre-draft player rankings
## Troubleshooting
### "Failed to connect to MCP server"
- Verify Python path in Claude Desktop config
- Ensure all Python dependencies are installed
- Check that file paths are absolute, not relative
### "Token expired" errors
- Run `python refresh_yahoo_token.py`
- Restart Claude Desktop after refreshing
### "No leagues found"
- Verify you have active leagues for the current season
- Check that YAHOO_GUID is set correctly in `.env`
- Ensure your Yahoo account has fantasy leagues
### "Cannot find team"
- Make sure YAHOO_GUID is set in both `.env` and Claude config
- Verify you're a member of the leagues
### Python Import Errors
- Ensure all requirements are installed: `pip install -r requirements.txt`
- If using virtual environment, make sure it's activated
## Testing Your Installation
Run the test suite to verify everything is working:
```bash
# Test league discovery
python test_all_leagues.py
# Test team name retrieval
python test_team_names.py
# Test waiver wire and rankings
python test_waiver_draft.py
```
## Updating
To get the latest updates:
```bash
git pull origin main
pip install -r requirements.txt --upgrade
```
Then restart Claude Desktop.
## Support
For issues or questions:
1. Check the [GitHub repository](https://github.com/derekrbreese/fantasy-football-mcp)
2. Review the CLAUDE.md file for development details
3. Ensure your Yahoo tokens are current
## Security Notes
- Never share your Yahoo API credentials
- The `.env` file contains sensitive tokens
- This repository should remain private
- Tokens expire after 1 hour (auto-refresh available)
- Refresh tokens last ~60 days if used regularly
```
--------------------------------------------------------------------------------
/docs/WAIVER_WIRE_VALIDATION_FIX.md:
--------------------------------------------------------------------------------
```markdown
# Waiver Wire Validation Fix
**Date**: October 2, 2025
**Commit**: 96dd1c5
**Files Modified**:
- `src/handlers/player_handlers.py`
- `fastmcp_server.py`
## Issue Summary
The `ff_get_waiver_wire` handler had validation issues that could cause errors when parameters were missing, null, or invalid. This affected both the MCP legacy handler and the FastMCP wrapper.
## Problems Fixed
### 1. Missing Parameter Validation
**Before**: Basic check for `league_key` with minimal error message
```python
if not arguments.get("league_key"):
return {"error": "league_key is required"}
```
**After**: Structured error response with clear messaging
```python
if not arguments.get("league_key"):
return {
"status": "error",
"error": "league_key is required",
"message": "Please provide a league_key parameter"
}
```
### 2. Null Position Parameter Handling
**Before**: Position could be `None` causing downstream issues
```python
position = arguments.get("position", "all")
# No null check
```
**After**: Explicit null-to-default conversion
```python
position = arguments.get("position", "all")
if position is None:
position = "all"
```
### 3. Sort Parameter Validation
**Before**: No validation, accepted any value
```python
sort = arguments.get("sort", "rank")
```
**After**: Validates against allowed values
```python
sort = arguments.get("sort", "rank")
if sort not in ["rank", "points", "owned", "trending"]:
sort = "rank"
```
### 4. Count Parameter Sanitization
**Before**: No type checking or bounds validation
```python
count = arguments.get("count", 30)
```
**After**: Type conversion and bounds checking
```python
count = arguments.get("count", 30)
try:
count = int(count)
if count < 1:
count = 30
except (ValueError, TypeError):
count = 30
```
### 5. Empty Result Handling
**Before**: Inconsistent response format
```python
if not basic_players:
return {
"league_key": league_key,
"message": "No available players found or error retrieving data",
}
```
**After**: Consistent success response with empty data
```python
if not basic_players:
return {
"status": "success",
"league_key": league_key,
"position": position,
"sort": sort,
"total_players": 0,
"players": [],
"message": "No available players found matching the criteria",
}
```
### 6. FastMCP Wrapper Issues
**Before**: Missing parameters in function signature
```python
async def ff_get_waiver_wire(
ctx: Context,
league_key: str,
position: Optional[str] = None,
sort: Literal["rank", "points", "owned", "trending"] = "rank",
count: int = 30,
include_expert_analysis: bool = True,
data_level: Optional[Literal["basic", "standard", "full"]] = None,
) -> Dict[str, Any]:
```
**After**: Complete parameter set
```python
async def ff_get_waiver_wire(
ctx: Context,
league_key: str,
position: Optional[str] = None,
sort: Literal["rank", "points", "owned", "trending"] = "rank",
count: int = 30,
week: Optional[int] = None, # ADDED
team_key: Optional[str] = None, # ADDED
include_expert_analysis: bool = True,
data_level: Optional[Literal["basic", "standard", "full"]] = None,
) -> Dict[str, Any]:
```
**Before**: Missing parameters in legacy tool call
```python
result = await _call_legacy_tool(
"ff_get_waiver_wire",
ctx=ctx,
league_key=league_key,
position=position,
sort=sort,
count=count,
include_projections=include_projections,
include_external_data=include_external_data,
include_analysis=include_analysis,
)
```
**After**: All parameters passed through
```python
# Handle position default - convert None to "all"
if position is None:
position = "all"
result = await _call_legacy_tool(
"ff_get_waiver_wire",
ctx=ctx,
league_key=league_key,
position=position,
sort=sort,
count=count,
week=week, # ADDED
team_key=team_key, # ADDED
include_projections=include_projections,
include_external_data=include_external_data,
include_analysis=include_analysis,
)
```
## Testing Results
All validation scenarios now pass:
### Test 1: Valid Parameters ✅
```python
{
'league_key': '461.l.61410',
'position': 'QB',
'count': 5
}
# Result: Success - returns players
```
### Test 2: Null Position ✅
```python
{
'league_key': '461.l.61410',
'position': None, # Should default to 'all'
'count': 5
}
# Result: Success - position defaults to 'all'
```
### Test 3: Missing Required Parameter ✅
```python
{} # No league_key
# Result: Error with clear message
{
"status": "error",
"error": "league_key is required",
"message": "Please provide a league_key parameter"
}
```
## Benefits
### 1. **Robustness**
- Handles edge cases gracefully
- No crashes from invalid input
- Clear error messages for debugging
### 2. **Consistency**
- All responses have `status` field
- Empty results treated as success, not error
- Structured error format across handlers
### 3. **Completeness**
- FastMCP wrapper now supports all legacy parameters
- Week and team_key parameters properly passed through
- No data loss in parameter translation
### 4. **Developer Experience**
- Clear validation error messages
- Type coercion prevents type errors
- Fallback defaults for invalid values
## Impact
### User-Facing
- ✅ No more crashes from null/missing parameters
- ✅ Clear error messages when parameters are wrong
- ✅ Consistent response format for empty results
### Developer-Facing
- ✅ Easier debugging with structured errors
- ✅ FastMCP wrapper feature-complete
- ✅ Better code maintainability
## Backward Compatibility
✅ **Fully backward compatible**
- Default values match previous behavior
- No breaking changes to API contract
- Existing valid calls work identically
## Future Improvements
Potential enhancements for other handlers:
1. **Apply Similar Validation**
- Use this pattern in other player handlers
- Standardize error response format across all handlers
2. **Schema Validation**
- Consider using Pydantic models for validation
- Centralize validation logic
3. **Parameter Documentation**
- Add inline examples in docstrings
- Document valid values for enum parameters
## Files Changed
### `src/handlers/player_handlers.py` (Lines 237-300)
- Added comprehensive parameter validation
- Improved error messages
- Better null handling
- Type coercion for count
### `fastmcp_server.py` (Lines 536-625)
- Added missing week and team_key parameters
- Added null-to-default conversion for position
- Updated legacy tool call to pass all parameters
## Verification
Tested with:
- ✅ Valid parameters
- ✅ Null parameters
- ✅ Missing parameters
- ✅ Invalid parameter types
- ✅ Invalid parameter values
All tests pass successfully.
---
**Status**: ✅ Fixed and Deployed
**Commit**: 96dd1c5
**Pushed**: October 2, 2025
```
--------------------------------------------------------------------------------
/utils/verify_setup.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Verify Yahoo Fantasy Football MCP Setup
Checks credentials and configuration
"""
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
import json
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
def check_env_file():
"""Check if .env file exists and has required variables"""
print("\n1. Checking .env file...")
env_path = Path('.env')
if not env_path.exists():
print(" ❌ .env file not found!")
print(" Fix: Copy .env.example to .env")
return False
load_dotenv()
required_vars = {
'YAHOO_CLIENT_ID': 'Your app client ID from Yahoo',
'YAHOO_CLIENT_SECRET': 'Your app client secret from Yahoo',
}
optional_vars = {
'YAHOO_ACCESS_TOKEN': 'OAuth access token (from setup_yahoo_auth.py)',
'YAHOO_REFRESH_TOKEN': 'OAuth refresh token (from setup_yahoo_auth.py)',
'YAHOO_GUID': 'Your Yahoo user ID (from setup_yahoo_auth.py)',
}
missing_required = []
missing_optional = []
print(" ✅ .env file found")
print("\n Required credentials:")
for var, desc in required_vars.items():
value = os.getenv(var)
if not value or value.startswith('your_') or value.startswith('YOUR_'):
print(f" ❌ {var}: Not configured")
missing_required.append(var)
else:
print(f" ✅ {var}: Configured ({len(value)} chars)")
print("\n OAuth credentials (from authentication):")
for var, desc in optional_vars.items():
value = os.getenv(var)
if not value:
print(f" ⚠️ {var}: Not set (run setup_yahoo_auth.py)")
missing_optional.append(var)
else:
# Show partial value for verification
if var == 'YAHOO_GUID':
print(f" ✅ {var}: {value}")
else:
print(f" ✅ {var}: {value[:20]}... ({len(value)} chars)")
if missing_required:
print(f"\n ❌ Missing required credentials: {', '.join(missing_required)}")
print(" Fix: Get these from https://developer.yahoo.com/apps/")
return False
if missing_optional:
print(f"\n ⚠️ Missing OAuth tokens. Run: python utils/setup_yahoo_auth.py")
return True
def check_yahoo_credentials():
"""Verify Yahoo credentials format"""
print("\n2. Verifying credential format...")
client_id = os.getenv('YAHOO_CLIENT_ID', '')
client_secret = os.getenv('YAHOO_CLIENT_SECRET', '')
issues = []
# Check Client ID format
if client_id:
if len(client_id) > 100:
issues.append("Client ID seems too long (might include extra parameters)")
print(" ⚠️ Client ID might be base64 encoded with parameters")
print(" Should look like: dj0yJmk9XXXXXXXXX")
elif client_id.startswith('dj0yJmk9'):
print(" ✅ Client ID format looks correct")
else:
print(" ⚠️ Client ID format might be incorrect")
# Check Client Secret format
if client_secret:
if len(client_secret) == 40 and all(c in '0123456789abcdef' for c in client_secret.lower()):
print(" ✅ Client Secret format looks correct (40-char hex)")
else:
print(f" ⚠️ Client Secret format might be incorrect (length: {len(client_secret)})")
issues.append("Client Secret should be a 40-character hexadecimal string")
return len(issues) == 0
def check_dependencies():
"""Check if required Python packages are installed"""
print("\n3. Checking Python dependencies...")
required_packages = [
'mcp',
'aiohttp',
'pydantic',
'dotenv',
'requests'
]
missing = []
for package in required_packages:
try:
__import__(package.replace('-', '_'))
print(f" ✅ {package}: Installed")
except ImportError:
print(f" ❌ {package}: Not installed")
missing.append(package)
if missing:
print(f"\n ❌ Missing packages: {', '.join(missing)}")
print(" Fix: pip install -r requirements.txt")
return False
return True
def check_claude_config():
"""Check if Claude Desktop config exists"""
print("\n4. Checking Claude Desktop configuration...")
import platform
system = platform.system()
if system == 'Darwin': # macOS
config_path = Path.home() / 'Library' / 'Application Support' / 'Claude' / 'claude_desktop_config.json'
elif system == 'Windows':
config_path = Path(os.environ['APPDATA']) / 'Claude' / 'claude_desktop_config.json'
else: # Linux
config_path = Path.home() / '.config' / 'Claude' / 'claude_desktop_config.json'
if config_path.exists():
print(f" ✅ Claude config found: {config_path}")
try:
with open(config_path) as f:
config = json.load(f)
if 'mcpServers' in config and 'fantasy-football' in config['mcpServers']:
print(" ✅ Fantasy Football MCP server configured")
else:
print(" ⚠️ Fantasy Football MCP not configured in Claude")
print(" Fix: Add configuration to claude_desktop_config.json")
print(" See INSTALLATION.md for details")
except json.JSONDecodeError:
print(" ❌ Claude config file is invalid JSON")
else:
print(f" ⚠️ Claude config not found at: {config_path}")
print(" This is OK if Claude Desktop isn't installed yet")
return True
def main():
"""Run all verification checks"""
print("=" * 60)
print("Yahoo Fantasy Football MCP - Setup Verification")
print("=" * 60)
all_good = True
# Check .env file
if not check_env_file():
all_good = False
# Check credential format
if not check_yahoo_credentials():
all_good = False
# Check dependencies
if not check_dependencies():
all_good = False
# Check Claude config
check_claude_config()
print("\n" + "=" * 60)
if all_good:
print("✅ Setup looks good!")
print("\nNext steps:")
if not os.getenv('YAHOO_ACCESS_TOKEN'):
print("1. Run: python utils/setup_yahoo_auth.py")
print("2. Configure Claude Desktop (see INSTALLATION.md)")
else:
print("1. Configure Claude Desktop if not done")
print("2. Restart Claude Desktop")
print("3. Test with: 'Show me my fantasy football leagues'")
else:
print("❌ Setup needs attention")
print("\nPlease fix the issues above and run this script again.")
print("See YAHOO_SETUP.md for detailed instructions.")
print("=" * 60)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/handlers/matchup_handlers.py:
--------------------------------------------------------------------------------
```python
"""Matchup MCP tool handlers."""
from typing import Any
# These will be injected from main file
get_user_team_key = None
get_user_team_info = None
yahoo_api_call = None
parse_team_roster = None
async def handle_ff_get_matchup(arguments: dict) -> dict:
"""Get matchup information for a team in a specific week.
Args:
arguments: Dict containing:
- league_key: League identifier
- week: Week number (optional, defaults to current)
Returns:
Dict with matchup data
"""
league_key = arguments.get("league_key")
week = arguments.get("week")
team_key = await get_user_team_key(league_key)
if not team_key:
return {"error": f"Could not find your team in league {league_key}"}
week_param = f";week={week}" if week else ""
data = await yahoo_api_call(f"team/{team_key}/matchups{week_param}")
return {
"league_key": league_key,
"team_key": team_key,
"week": week or "current",
"message": "Matchup data retrieved",
"raw_matchups": data,
}
async def handle_ff_compare_teams(arguments: dict) -> dict:
"""Compare rosters of two teams.
Args:
arguments: Dict containing:
- league_key: League identifier
- team_key_a: First team identifier
- team_key_b: Second team identifier
Returns:
Dict with comparison data
"""
league_key = arguments.get("league_key")
team_key_a = arguments.get("team_key_a")
team_key_b = arguments.get("team_key_b")
data_a = await yahoo_api_call(f"team/{team_key_a}/roster")
data_b = await yahoo_api_call(f"team/{team_key_b}/roster")
roster_a = parse_team_roster(data_a)
roster_b = parse_team_roster(data_b)
return {
"league_key": league_key,
"team_a": {"team_key": team_key_a, "roster": roster_a},
"team_b": {"team_key": team_key_b, "roster": roster_b},
}
async def handle_ff_build_lineup(arguments: dict) -> dict:
"""Build optimal lineup using advanced analytics.
Args:
arguments: Dict containing:
- league_key: League identifier
- week: Week number (optional)
- strategy: "balanced", "floor", or "ceiling" (default: "balanced")
- use_llm: Use LLM for additional insights (default: False)
Returns:
Dict with optimal lineup and recommendations
"""
league_key = arguments.get("league_key")
week = arguments.get("week")
strategy = arguments.get("strategy", "balanced")
use_llm = arguments.get("use_llm", False)
team_key = await get_user_team_key(league_key)
if not team_key:
return {"error": f"Could not find your team in league {league_key}"}
try:
roster_data = await yahoo_api_call(f"team/{team_key}/roster")
try:
from lineup_optimizer import lineup_optimizer
except ImportError as exc:
return {
"error": f"Lineup optimizer unavailable: {exc}",
"suggestion": "Please check lineup_optimizer.py dependencies",
"league_key": league_key,
"team_key": team_key,
}
players = await lineup_optimizer.parse_yahoo_roster(roster_data)
if not players:
return {
"error": "Failed to parse Yahoo roster data",
"league_key": league_key,
"team_key": team_key,
"suggestion": "Check roster data format or try refreshing",
}
players = await lineup_optimizer.enhance_with_external_data(players, week=week)
optimization = await lineup_optimizer.optimize_lineup_smart(
players,
strategy,
week,
use_llm,
)
if optimization["status"] == "error":
return {
"status": "error",
"error": "Lineup optimization failed",
"league_key": league_key,
"team_key": team_key,
"errors": optimization.get("errors", []),
"details": optimization.get("errors", []),
"data_quality": optimization.get("data_quality", {}),
}
starters_formatted = {}
for pos, player in optimization["starters"].items():
starters_formatted[pos] = {
"name": player.name,
"tier": player.player_tier.upper() if player.player_tier else "UNKNOWN",
"team": player.team,
"opponent": player.opponent,
"matchup_score": player.matchup_score,
"matchup": player.matchup_description,
"composite_score": round(player.composite_score, 1),
"yahoo_proj": (
round(player.yahoo_projection, 1) if player.yahoo_projection else None
),
"sleeper_proj": (
round(player.sleeper_projection, 1) if player.sleeper_projection else None
),
"trending": (
f"{player.trending_score:,} adds" if player.trending_score > 0 else None
),
"floor": round(player.floor_projection, 1) if player.floor_projection else None,
"ceiling": (
round(player.ceiling_projection, 1) if player.ceiling_projection else None
),
}
bench_formatted = [
{
"name": player.name,
"position": player.position,
"opponent": player.opponent,
"composite_score": round(player.composite_score, 1),
"matchup_score": player.matchup_score,
"tier": player.player_tier.upper() if player.player_tier else "UNKNOWN",
}
for player in optimization["bench"][:5]
]
result: dict[str, Any] = {
"status": optimization["status"],
"league_key": league_key,
"team_key": team_key,
"week": week or "current",
"strategy": strategy,
"optimal_lineup": starters_formatted,
"bench": bench_formatted,
"recommendations": optimization["recommendations"],
"errors": optimization.get("errors", []),
"analysis": {
"total_players": optimization["data_quality"]["total_players"],
"valid_players": optimization["data_quality"]["valid_players"],
"players_with_projections": optimization["data_quality"][
"players_with_projections"
],
"players_with_matchup_data": optimization["data_quality"][
"players_with_matchup_data"
],
"strategy_used": optimization["strategy_used"],
"data_sources": [
"Yahoo projections",
"Sleeper rankings",
"Matchup analysis",
"Trending data",
],
},
}
if optimization.get("errors"):
result["warnings"] = optimization["errors"]
return result
except Exception as exc:
return {
"error": f"Unexpected error during lineup optimization: {exc}",
"league_key": league_key,
"team_key": team_key,
"suggestion": "Try again or check system logs for details",
}
```
--------------------------------------------------------------------------------
/examples/demo_enhancement_layer.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Demonstration of the Player Enhancement Layer.
This script shows how the enhancement layer enriches player data with:
- Bye week detection and projection zeroing
- Recent performance stats and trends
- Performance flags (BREAKOUT, DECLINING, etc.)
- Adjusted projections based on recent reality
"""
import asyncio
from datetime import datetime
def print_player_comparison(name, before, after):
"""Print before/after comparison for a player."""
print(f"\n{'='*70}")
print(f"PLAYER: {name}")
print(f"{'='*70}")
print("\n📊 BEFORE Enhancement (Stale Sleeper Projection):")
print(f" Yahoo Projection: {before['yahoo_proj']:.1f} pts")
print(f" Sleeper Projection: {before['sleeper_proj']:.1f} pts")
print(f" Recommendation: {before['recommendation']}")
print("\n✨ AFTER Enhancement (Reality-Adjusted):")
print(f" Yahoo Projection: {after['yahoo_proj']:.1f} pts")
print(f" Sleeper Projection: {after['sleeper_proj']:.1f} pts")
print(f" Adjusted Projection: {after.get('adjusted_proj', 'N/A')}")
print(f" Recommendation: {after['recommendation']}")
if after.get("on_bye"):
print(f" 🚫 ON BYE: {after.get('context', 'N/A')}")
if after.get("flags"):
print(f" 🏷️ Flags: {', '.join(after['flags'])}")
if after.get("recent_stats"):
print(f" 📈 Recent: {after['recent_stats']}")
if after.get("context") and not after.get("on_bye"):
print(f" 💡 Context: {after['context']}")
async def demo_bye_week_detection():
"""Demonstrate bye week detection."""
print("\n" + "=" * 70)
print("SCENARIO 1: Bye Week Detection - Nico Collins (Week 6)")
print("=" * 70)
print("\nProblem: Player on bye still showing projections and 'Start' recommendation")
before = {"yahoo_proj": 14.5, "sleeper_proj": 15.2, "recommendation": "Start"}
after = {
"yahoo_proj": 0.0,
"sleeper_proj": 0.0,
"adjusted_proj": 0.0,
"recommendation": "BYE WEEK - DO NOT START",
"on_bye": True,
"flags": ["ON_BYE"],
"context": "Player is on bye Week 6",
}
print_player_comparison("Nico Collins (WR, HOU)", before, after)
print("\n✅ FIXED: Projections zeroed, clear 'DO NOT START' warning")
async def demo_breakout_detection():
"""Demonstrate breakout player detection."""
print("\n" + "=" * 70)
print("SCENARIO 2: Breakout Performance - Rico Dowdle (Week 5)")
print("=" * 70)
print("\nProblem: 206 yards + 2 TDs Week 5, became lead back, still projects 4.0")
before = {"yahoo_proj": 5.2, "sleeper_proj": 4.0, "recommendation": "Bench"}
after = {
"yahoo_proj": 5.2,
"sleeper_proj": 4.0,
"adjusted_proj": 14.8,
"recommendation": "Strong Start",
"on_bye": False,
"flags": ["BREAKOUT_CANDIDATE", "TRENDING_UP"],
"recent_stats": "L3W avg: 18.5 pts/game",
"context": "Recent breakout: averaging 18.5 pts over last 3 weeks (projection: 4.0)",
}
print_player_comparison("Rico Dowdle (RB, DAL)", before, after)
print("\n✅ FIXED: Adjusted projection reflects recent performance (4.0 → 14.8)")
print(" Performance flags alert users to breakout potential")
async def demo_declining_role():
"""Demonstrate declining player detection."""
print("\n" + "=" * 70)
print("SCENARIO 3: Declining Role - Travis Etienne (Recent Weeks)")
print("=" * 70)
print("\nProblem: Declining role/touches, still projects 7.7 pts")
before = {"yahoo_proj": 8.1, "sleeper_proj": 7.7, "recommendation": "Start"}
after = {
"yahoo_proj": 8.1,
"sleeper_proj": 7.7,
"adjusted_proj": 5.2,
"recommendation": "Bench/Consider",
"on_bye": False,
"flags": ["DECLINING_ROLE", "TRENDING_DOWN"],
"recent_stats": "L3W avg: 4.8 pts/game",
"context": "Declining role: averaging 4.8 pts over last 3 weeks (projection: 7.7)",
}
print_player_comparison("Travis Etienne (RB, JAX)", before, after)
print("\n✅ FIXED: Adjusted projection reflects declining role (7.7 → 5.2)")
print(" Flags warn users about performance decline")
async def demo_consistent_performer():
"""Demonstrate consistent performer."""
print("\n" + "=" * 70)
print("SCENARIO 4: Consistent Performer - Steady Production")
print("=" * 70)
before = {"yahoo_proj": 12.0, "sleeper_proj": 11.5, "recommendation": "Start"}
after = {
"yahoo_proj": 12.0,
"sleeper_proj": 11.5,
"adjusted_proj": 11.9,
"recommendation": "Start",
"on_bye": False,
"flags": ["CONSISTENT"],
"recent_stats": "L3W avg: 12.2 pts/game",
"context": "L3W avg: 12.2 pts",
}
print_player_comparison("Consistent Player (WR, KC)", before, after)
print("\n✅ Projection matches reality, CONSISTENT flag confirms reliability")
async def show_api_response_example():
"""Show what the enhanced API response looks like."""
print("\n" + "=" * 70)
print("ENHANCED API RESPONSE EXAMPLE")
print("=" * 70)
print(
"""
When you call ff_get_roster or ff_get_waiver_wire with enhancements enabled,
you now get these additional fields:
{
"players": [
{
"name": "Rico Dowdle",
"position": "RB",
"team": "DAL",
// Standard projections
"yahoo_projection": 5.2,
"sleeper_projection": 4.0,
// 🆕 ENHANCEMENT LAYER FIELDS
"bye_week": 7,
"on_bye": false,
"adjusted_projection": 14.8,
"performance_flags": ["BREAKOUT_CANDIDATE", "TRENDING_UP"],
"enhancement_context": "Recent breakout: averaging 18.5 pts over last 3 weeks (projection: 4.0)",
// Analysis includes recent performance
"roster_analysis": {
"start_recommendation": "Strong Start",
"recent_performance": "L3W avg: 18.5 pts/game",
"trend": "IMPROVING",
"performance_alerts": ["BREAKOUT_CANDIDATE", "TRENDING_UP"]
}
}
]
}
"""
)
async def main():
"""Run all demonstrations."""
print("=" * 70)
print("PLAYER ENHANCEMENT LAYER - DEMONSTRATION")
print("=" * 70)
print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("\nThis demonstrates how the enhancement layer fixes projection issues:")
await demo_bye_week_detection()
await demo_breakout_detection()
await demo_declining_role()
await demo_consistent_performer()
await show_api_response_example()
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
print(
"""
The Player Enhancement Layer successfully addresses all reported issues:
✅ Bye Week Detection
- Zeros out projections for players on bye
- Sets clear "BYE WEEK - DO NOT START" recommendation
- Prevents users from accidentally starting bye week players
✅ Breakout Performance Recognition
- Fetches last 1-3 weeks actual stats from Sleeper API
- Adjusts projections upward when recent performance exceeds projection
- Flags players as "BREAKOUT_CANDIDATE" or "TRENDING_UP"
✅ Declining Role Detection
- Adjusts projections downward when recent performance < 70% of projection
- Flags players as "DECLINING_ROLE" or "UNDERPERFORMING"
- Helps users avoid starting players losing touches/opportunities
✅ Recent Performance Context
- Shows "L3W avg: X.X pts/game" for informed decisions
- Trend indicators: IMPROVING, DECLINING, STABLE
- Performance flags provide at-a-glance insights
The layer integrates seamlessly with existing tools:
- ff_get_roster
- ff_get_waiver_wire
- ff_get_players
- ff_build_lineup
All enhancements are non-breaking - existing functionality continues to work
while users get richer, more accurate data for better decisions.
"""
)
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/tests/test_real_data.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Test enhancement layer with real Yahoo Fantasy data."""
import asyncio
import os
import sys
from datetime import datetime
async def test_with_real_roster():
"""Test enhancement layer with real roster from Yahoo."""
print("=" * 60)
print("REAL DATA TEST: Roster Enhancement")
print("=" * 60)
# Check for environment variables
required_vars = [
"YAHOO_CONSUMER_KEY",
"YAHOO_CONSUMER_SECRET",
"YAHOO_ACCESS_TOKEN",
"YAHOO_REFRESH_TOKEN",
]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
print(f"⚠️ SKIP: Missing environment variables: {missing}")
print("This test requires Yahoo API credentials")
return True
try:
# Import after env check
import sys
sys.path.insert(0, "/workspaces/fantasy-football-mcp-server")
from src.handlers.roster_handlers import handle_ff_get_roster
from fantasy_football_multi_league import (
discover_leagues,
get_user_team_info,
yahoo_api_call,
)
from src.parsers.yahoo_parsers import parse_team_roster
# Discover leagues
print("\n1. Discovering leagues...")
leagues = await discover_leagues()
if not leagues:
print("⚠️ No leagues found")
return True
print(f"Found {len(leagues)} league(s):")
for league in leagues[:3]: # Show first 3
print(f" - {league.get('name')} ({league.get('league_key')})")
# Use first league
league_key = leagues[0].get("league_key")
print(f"\n2. Testing with league: {league_key}")
# Inject dependencies for handlers
from src.handlers import (
inject_roster_dependencies,
)
inject_roster_dependencies(
get_user_team_info=get_user_team_info,
yahoo_api_call=yahoo_api_call,
parse_team_roster=parse_team_roster,
)
# Get roster with enhancements
print("\n3. Fetching roster with enhancements...")
result = await handle_ff_get_roster(
{
"league_key": league_key,
"data_level": "enhanced", # This triggers enhancement
"include_projections": True,
"include_external_data": True,
"include_analysis": True,
}
)
if result.get("status") != "success":
print(f"❌ FAIL: {result.get('error', 'Unknown error')}")
return False
# Analyze results
print(f"\n4. Analyzing enhancement results...")
all_players = result.get("all_players", [])
print(f"Total players: {len(all_players)}")
# Check for bye week players
bye_week_players = [p for p in all_players if p.get("on_bye")]
print(f"\n✓ Players on bye this week: {len(bye_week_players)}")
for player in bye_week_players[:5]: # Show first 5
print(
f" - {player.get('name')} ({player.get('position')}) - Week {player.get('bye_week')}"
)
print(f" Yahoo proj: {player.get('yahoo_projection', 0):.1f}")
print(f" Sleeper proj: {player.get('sleeper_projection', 0):.1f}")
print(f" Context: {player.get('enhancement_context', 'N/A')}")
# Check for players with performance flags
flagged_players = [p for p in all_players if p.get("performance_flags")]
print(f"\n✓ Players with performance flags: {len(flagged_players)}")
for player in flagged_players[:5]: # Show first 5
flags = player.get("performance_flags", [])
print(f" - {player.get('name')} ({player.get('position')}): {', '.join(flags)}")
if player.get("enhancement_context"):
print(f" {player.get('enhancement_context')}")
# Check for adjusted projections
adjusted_players = [p for p in all_players if p.get("adjusted_projection") is not None]
print(f"\n✓ Players with adjusted projections: {len(adjusted_players)}")
# Show some examples of adjustments
for player in adjusted_players[:5]:
sleeper = player.get("sleeper_projection", 0)
adjusted = player.get("adjusted_projection", 0)
if sleeper > 0:
diff = adjusted - sleeper
print(f" - {player.get('name')}: {sleeper:.1f} → {adjusted:.1f} ({diff:+.1f})")
print("\n✅ PASS: Enhancement layer successfully integrated with real data")
return True
except Exception as e:
print(f"❌ ERROR: {e}")
import traceback
traceback.print_exc()
return False
async def test_waiver_wire_enhancements():
"""Test enhancement with waiver wire players."""
print("\n" + "=" * 60)
print("REAL DATA TEST: Waiver Wire Enhancement")
print("=" * 60)
required_vars = [
"YAHOO_CONSUMER_KEY",
"YAHOO_CONSUMER_SECRET",
"YAHOO_ACCESS_TOKEN",
"YAHOO_REFRESH_TOKEN",
]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
print(f"⚠️ SKIP: Missing environment variables")
return True
try:
import sys
sys.path.insert(0, "/workspaces/fantasy-football-mcp-server")
from src.handlers.player_handlers import handle_ff_get_waiver_wire
from fantasy_football_multi_league import (
discover_leagues,
get_waiver_wire_players,
yahoo_api_call,
)
# Get league
leagues = await discover_leagues()
if not leagues:
print("⚠️ No leagues found")
return True
league_key = leagues[0].get("league_key")
print(f"Testing with league: {league_key}")
# Inject dependencies
from src.handlers import inject_player_dependencies
inject_player_dependencies(
yahoo_api_call=yahoo_api_call,
get_waiver_wire_players=get_waiver_wire_players,
)
# Get waiver wire with enhancements
print("\nFetching top waiver wire RBs with enhancements...")
result = await handle_ff_get_waiver_wire(
{
"league_key": league_key,
"position": "RB",
"count": 10,
"include_projections": True,
"include_external_data": True,
"include_analysis": True,
}
)
if result.get("status") != "success":
print(f"❌ FAIL: {result.get('error')}")
return False
players = result.get("enhanced_players", [])
print(f"\nFound {len(players)} RBs on waiver wire")
# Check enhancements
bye_players = [p for p in players if p.get("on_bye")]
print(f"\n✓ RBs on bye: {len(bye_players)}")
for player in bye_players[:3]:
print(f" - {player['name']}: {player.get('enhancement_context', 'N/A')}")
flagged = [p for p in players if p.get("performance_flags")]
print(f"\n✓ RBs with performance flags: {len(flagged)}")
for player in flagged[:3]:
print(f" - {player['name']}: {', '.join(player.get('performance_flags', []))}")
print("\n✅ PASS: Waiver wire enhancements working")
return True
except Exception as e:
print(f"❌ ERROR: {e}")
import traceback
traceback.print_exc()
return False
async def main():
print("=" * 60)
print("REAL DATA ENHANCEMENT TESTS")
print("=" * 60)
print(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
results = []
results.append(("Roster Enhancement", await test_with_real_roster()))
results.append(("Waiver Wire Enhancement", await test_waiver_wire_enhancements()))
# Summary
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
passed = sum(1 for _, r in results if r)
failed = sum(1 for _, r in results if not r)
for name, result in results:
status = "✅ PASS" if result else "❌ FAIL"
print(f"{status}: {name}")
print(f"\nTotal: {passed}/{len(results)} tests passed")
return 0 if failed == 0 else 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
```
--------------------------------------------------------------------------------
/utils/reauth_yahoo.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Re-authenticate with Yahoo Fantasy Sports
Full OAuth2 flow when refresh token expires
"""
import os
import json
import time
import webbrowser
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import requests
from dotenv import load_dotenv
import threading
# Load environment
load_dotenv()
# Global to store the auth code
auth_code = None
class CallbackHandler(BaseHTTPRequestHandler):
"""Handle OAuth callback."""
def do_GET(self):
global auth_code
# Parse the callback URL
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
if "code" in params:
auth_code = params["code"][0]
# Send success response
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
success_html = """
<html>
<head><title>Success!</title></head>
<body style="font-family: Arial; text-align: center; padding: 50px;">
<h1 style="color: green;">✅ Authentication Successful!</h1>
<p>You can close this window and return to the terminal.</p>
<script>window.setTimeout(function(){window.close();}, 3000);</script>
</body>
</html>
"""
self.wfile.write(success_html.encode())
else:
# Error response
self.send_response(400)
self.send_header("Content-type", "text/html")
self.end_headers()
error_html = """
<html>
<head><title>Error</title></head>
<body style="font-family: Arial; text-align: center; padding: 50px;">
<h1 style="color: red;">❌ Authentication Failed</h1>
<p>No authorization code received.</p>
</body>
</html>
"""
self.wfile.write(error_html.encode())
def log_message(self, format, *args):
# Suppress default logging
pass
def run_callback_server(port=8000):
"""Run the callback server in a thread."""
server = HTTPServer(("localhost", port), CallbackHandler)
server.timeout = 60 # 60 second timeout
server.handle_request() # Handle one request then stop
def reauth_yahoo():
"""Complete re-authentication flow."""
print("=" * 60)
print("Yahoo Fantasy Sports Re-Authentication")
print("=" * 60)
print()
# Get credentials
client_id = os.getenv("YAHOO_CONSUMER_KEY")
client_secret = os.getenv("YAHOO_CONSUMER_SECRET")
if not client_id or not client_secret:
print("❌ Missing YAHOO_CONSUMER_KEY or YAHOO_CONSUMER_SECRET in .env")
return False
# OAuth URLs
auth_url = "https://api.login.yahoo.com/oauth2/request_auth"
token_url = "https://api.login.yahoo.com/oauth2/get_token"
# Callback configuration
callback_port = 8000
redirect_uri = f"http://localhost:{callback_port}/callback"
print(f"📌 Using redirect URI: {redirect_uri}")
print()
print("⚠️ IMPORTANT: Make sure this matches your Yahoo App settings!")
print()
# Start callback server in background
print(f"🌐 Starting callback server on port {callback_port}...")
server_thread = threading.Thread(target=run_callback_server, args=(callback_port,))
server_thread.daemon = True
server_thread.start()
# Build authorization URL
auth_params = {
"client_id": client_id,
"redirect_uri": redirect_uri,
"response_type": "code",
"language": "en-us",
}
auth_url_full = auth_url + "?" + "&".join([f"{k}={v}" for k, v in auth_params.items()])
print("🔗 Opening browser for Yahoo login...")
print()
print("If browser doesn't open, manually visit:")
print(auth_url_full)
print()
# Open browser
webbrowser.open(auth_url_full)
# Wait for callback
print("⏳ Waiting for authorization (60 seconds timeout)...")
# Wait for the server thread to complete
server_thread.join(timeout=65)
global auth_code
if not auth_code:
print("❌ No authorization code received. Timeout or user cancelled.")
return False
print(f"✅ Authorization code received!")
print()
# Exchange code for tokens
print("🔄 Exchanging code for tokens...")
token_data = {
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
"code": auth_code,
"grant_type": "authorization_code",
}
try:
response = requests.post(token_url, data=token_data)
if response.status_code == 200:
tokens = response.json()
access_token = tokens.get("access_token")
refresh_token = tokens.get("refresh_token")
expires_in = tokens.get("expires_in", 3600)
print("✅ Tokens received successfully!")
print(f" Token expires in: {expires_in} seconds ({expires_in/3600:.1f} hours)")
print()
# Get user GUID
guid = get_user_guid(access_token)
# Save to files
save_tokens(access_token, refresh_token, guid)
print("✅ Authentication complete!")
print()
print("📝 Tokens saved to:")
print(" - .env file")
print(" - claude_desktop_config.json")
print()
print("⚠️ IMPORTANT: Restart Claude Desktop to use the new tokens")
return True
else:
print(f"❌ Failed to get tokens: {response.status_code}")
print(f" Response: {response.text}")
return False
except Exception as e:
print(f"❌ Error getting tokens: {e}")
return False
def get_user_guid(access_token):
"""Get the user's Yahoo GUID."""
url = "https://fantasysports.yahooapis.com/fantasy/v2/users;use_login=1?format=json"
headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"}
try:
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
# Navigate the response to find GUID
users = data.get("fantasy_content", {}).get("users", {})
if "0" in users:
user = users["0"]["user"]
if isinstance(user, list) and len(user) > 0:
guid = user[0].get("guid")
if guid:
print(f"📌 Found user GUID: {guid}")
return guid
except:
pass
return None
def save_tokens(access_token, refresh_token, guid=None):
"""Save tokens to .env and Claude config."""
# Update .env
env_lines = []
env_path = ".env"
# Read existing or create new
if os.path.exists(env_path):
with open(env_path, "r") as f:
for line in f:
if not line.startswith(
("YAHOO_ACCESS_TOKEN=", "YAHOO_REFRESH_TOKEN=", "YAHOO_GUID=")
):
env_lines.append(line)
# Add new tokens
env_lines.append(f"YAHOO_ACCESS_TOKEN={access_token}\n")
env_lines.append(f"YAHOO_REFRESH_TOKEN={refresh_token}\n")
if guid:
env_lines.append(f"YAHOO_GUID={guid}\n")
with open(env_path, "w") as f:
f.writelines(env_lines)
# Update Claude config
config_path = "claude_desktop_config.json"
if os.path.exists(config_path):
try:
with open(config_path, "r") as f:
config = json.load(f)
if "mcpServers" in config and "fantasy-football" in config["mcpServers"]:
if "env" not in config["mcpServers"]["fantasy-football"]:
config["mcpServers"]["fantasy-football"]["env"] = {}
config["mcpServers"]["fantasy-football"]["env"]["YAHOO_ACCESS_TOKEN"] = access_token
config["mcpServers"]["fantasy-football"]["env"][
"YAHOO_REFRESH_TOKEN"
] = refresh_token
if guid:
config["mcpServers"]["fantasy-football"]["env"]["YAHOO_GUID"] = guid
with open(config_path, "w") as f:
json.dump(config, f, indent=4)
except:
pass
if __name__ == "__main__":
reauth_yahoo()
```
--------------------------------------------------------------------------------
/src/models/draft.py:
--------------------------------------------------------------------------------
```python
"""
Draft-specific data models for the Fantasy Football MCP Draft Assistant.
This module defines all data structures needed for draft evaluation and recommendations,
including draft state, pick evaluations, and recommendation objects.
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Any, Union
from pydantic import BaseModel, Field
from .player import Player, Position
class DraftStrategy(str, Enum):
"""Available draft strategies."""
CONSERVATIVE = "conservative"
AGGRESSIVE = "aggressive"
BALANCED = "balanced"
class PositionalNeed(str, Enum):
"""Positional need levels for roster construction."""
CRITICAL = "critical" # Zero players at position
HIGH = "high" # Below optimal roster construction
MEDIUM = "medium" # At optimal level
LOW = "low" # Above optimal level
SATURATED = "saturated" # Excess players
class DraftTier(int, Enum):
"""Player tiers for draft evaluation."""
ELITE = 1 # Tier 1 - Elite players
STUD = 2 # Tier 2 - Stud players
SOLID = 3 # Tier 3 - Solid players
FLEX = 4 # Tier 4 - Flex/depth players
BENCH = 5 # Tier 5 - Bench/flyer players
@dataclass
class DraftPosition:
"""Represents a draft position and round information."""
overall_pick: int
round_number: int
pick_in_round: int
picks_until_next: int
is_snake_draft: bool = True
@dataclass
class RosterNeed:
"""Represents positional needs for roster construction."""
position: Position
need_level: PositionalNeed
current_count: int
optimal_count: int
starter_slots: int
bye_week_conflicts: int = 0
@dataclass
class PlayerEvaluation:
"""Complete evaluation of a player for draft purposes."""
player: Player
overall_score: float
vorp_score: float
scarcity_score: float
need_score: float
bye_week_score: float
risk_score: float
upside_score: float
tier: DraftTier
adp: Optional[float] = None
projected_points: Optional[float] = None
replacement_level: Optional[float] = None
injury_risk: Optional[float] = None
consistency_score: Optional[float] = None
@dataclass
class OpportunityCost:
"""Analysis of opportunity cost for waiting vs taking a player."""
player: Player
survival_probability: float
cost_of_waiting: float
expected_value_next_round: float
recommendation: str # "take_now", "can_wait", "risky_wait"
@dataclass
class PositionalRun:
"""Detection and analysis of positional runs."""
position: Position
recent_picks: int
is_hot_run: bool
run_intensity: str # "emerging", "hot", "cooling"
recommendation: str
class DraftRecommendation(BaseModel):
"""Main recommendation object returned by draft evaluator."""
player: Dict[str, Any] # Player dict to avoid circular imports
overall_score: float = Field(..., description="Combined evaluation score (0-100)")
rank: int = Field(..., description="Ranking among available players")
tier: DraftTier = Field(..., description="Player tier classification")
# Score breakdowns
vorp_score: float = Field(..., description="Value Over Replacement Player score")
scarcity_score: float = Field(..., description="Positional scarcity score")
need_score: float = Field(..., description="Roster need score")
bye_week_score: float = Field(..., description="Bye week distribution score")
risk_score: float = Field(..., description="Injury/performance risk score")
upside_score: float = Field(..., description="Ceiling/upside potential score")
# Contextual information
projected_points: Optional[float] = Field(None, description="Season projection")
adp: Optional[float] = Field(None, description="Average Draft Position")
position_rank: Optional[int] = Field(None, description="Rank within position")
# Analysis
reasoning: str = Field(..., description="Human-readable explanation")
opportunity_cost: Optional[Dict[str, Any]] = Field(None, description="Wait vs take analysis")
positional_context: Optional[str] = Field(None, description="Position-specific insights")
class Config:
use_enum_values = True
class DraftState(BaseModel):
"""Current state of the draft and roster."""
league_key: str = Field(..., description="Yahoo league identifier")
draft_position: DraftPosition = Field(..., description="Current draft position")
current_roster: List[Dict[str, Any]] = Field(
default_factory=list, description="Currently drafted players"
)
available_players: List[Dict[str, Any]] = Field(
default_factory=list, description="Remaining available players"
)
# Roster analysis
roster_needs: List[RosterNeed] = Field(
default_factory=list, description="Positional needs assessment"
)
bye_week_distribution: Dict[int, int] = Field(
default_factory=dict, description="Bye weeks by count"
)
# Draft context
total_rounds: int = Field(default=16, description="Total draft rounds")
picks_remaining: int = Field(..., description="Remaining picks for user")
strategy: DraftStrategy = Field(
default=DraftStrategy.BALANCED, description="Selected draft strategy"
)
# Analysis flags
positional_runs: List[PositionalRun] = Field(
default_factory=list, description="Detected positional runs"
)
draft_phase: str = Field(..., description="early, middle, or late draft phase")
class DraftAnalysis(BaseModel):
"""Comprehensive analysis of the current draft situation."""
draft_state: DraftState
top_recommendations: List[DraftRecommendation] = Field(
..., description="Top N recommended picks"
)
# Strategic insights
key_insights: List[str] = Field(default_factory=list, description="Important strategic notes")
positional_priorities: Dict[str, float] = Field(
default_factory=dict, description="Position priority scores"
)
risk_factors: List[str] = Field(default_factory=list, description="Potential concerns")
# Context
analysis_timestamp: datetime = Field(default_factory=datetime.now)
strategy_weights: Dict[str, float] = Field(
default_factory=dict, description="Applied strategy weights"
)
class Config:
use_enum_values = True
@dataclass
class StrategyWeights:
"""Weights for different factors in draft evaluation."""
vorp: float = 0.30
scarcity: float = 0.25
need: float = 0.20
bye_week: float = 0.10
risk: float = 0.10
upside: float = 0.05
def __post_init__(self):
"""Validate weights sum to 1.0."""
total = sum([self.vorp, self.scarcity, self.need, self.bye_week, self.risk, self.upside])
if abs(total - 1.0) > 0.01:
raise ValueError(f"Strategy weights must sum to 1.0, got {total}")
# Predefined strategy weight configurations
STRATEGY_WEIGHTS = {
DraftStrategy.CONSERVATIVE: StrategyWeights(
vorp=0.35, scarcity=0.25, need=0.25, bye_week=0.10, risk=0.05, upside=0.0
),
DraftStrategy.AGGRESSIVE: StrategyWeights(
vorp=0.25, scarcity=0.20, need=0.15, bye_week=0.05, risk=-0.05, upside=0.40
),
DraftStrategy.BALANCED: StrategyWeights(
vorp=0.30, scarcity=0.25, need=0.20, bye_week=0.10, risk=0.10, upside=0.05
),
}
# Position requirements for standard roster construction
STANDARD_ROSTER_REQUIREMENTS = {
Position.QB: {"starters": 1, "optimal_total": 2, "max_useful": 3},
Position.RB: {"starters": 2, "optimal_total": 5, "max_useful": 7},
Position.WR: {"starters": 2, "optimal_total": 5, "max_useful": 7},
Position.TE: {"starters": 1, "optimal_total": 2, "max_useful": 3},
Position.K: {"starters": 1, "optimal_total": 1, "max_useful": 2},
Position.DEF: {"starters": 1, "optimal_total": 1, "max_useful": 2},
}
# Flex position can be filled by RB/WR/TE
FLEX_POSITIONS = [Position.RB, Position.WR, Position.TE]
# Injury risk multipliers by position
POSITION_INJURY_RISK = {
Position.QB: 1.0,
Position.RB: 1.5, # Higher injury risk
Position.WR: 1.2,
Position.TE: 1.1,
Position.K: 0.5, # Lower injury risk
Position.DEF: 0.5,
}
# Age-based risk adjustments
AGE_RISK_THRESHOLDS = {
Position.QB: 35,
Position.RB: 30,
Position.WR: 32,
Position.TE: 32,
Position.K: 40,
Position.DEF: None, # Not applicable
}
```
--------------------------------------------------------------------------------
/docs/LIVE_API_TESTING_SUMMARY.md:
--------------------------------------------------------------------------------
```markdown
# Live API Testing Summary - Phase 2b Verification
**Date**: October 2, 2025
**Branch**: `consolidate-fastmcp`
**Test Script**: `test_live_api.py`
## Executive Summary
Successfully executed comprehensive live API testing of all MCP tool handlers against the real Yahoo Fantasy Sports API. **All 22 tests passed with 100% success rate**, verifying that Phase 2b refactoring is production-ready.
## Test Execution
### Test Plan Implemented
- ✅ Created `test_live_api.py` - 650+ line comprehensive test script
- ✅ Organized tests by handler domain (Admin, League, Roster, Matchup, Player, Draft, Analytics)
- ✅ Implemented color-coded terminal output for real-time monitoring
- ✅ Automated results document generation (`LIVE_API_TEST_RESULTS.md`)
- ✅ Built-in safety features (rate limiting checks, delays between tests)
### Test Results
#### Overall Results
- **Total Tests**: 22
- **Passed**: 22 (100%)
- **Failed**: 0
- **Total Time**: 14.67 seconds
- **API Calls**: 22
- **Average Response Time**: 0.67 seconds
#### Results by Handler Domain
| Domain | Tests | Pass Rate | Notes |
|--------|-------|-----------|-------|
| Admin Handlers | 3/3 | 100% | Token refresh, cache, API status all working |
| League Handlers | 7/7 | 100% | All league discovery & info retrieval working |
| Roster Handlers | 3/3 | 100% | Basic, standard, and full roster modes verified |
| Matchup Handlers | 2/2 | 100% | Matchups and lineup optimization functional |
| Player Handlers | 2/2 | 100% | Player search and waiver wire analysis working |
| Draft Handlers | 4/4 | 100% | All draft tools operational |
| Analytics Handlers | 1/1 | 100% | Reddit sentiment analysis functional |
## Key Findings
### ✅ Successful Verifications
1. **Dependency Injection Working**
- All injected dependencies resolved correctly
- No circular dependency issues
- Handlers accessing required functions properly
2. **API Integration Intact**
- Yahoo Fantasy Sports API integration functional
- Token refresh mechanism working
- Rate limiting system operational
- Cache system functioning correctly
3. **External Services Working**
- Sleeper API integration operational (fallback projections active)
- Reddit API connection successful
- Multi-source data aggregation working
4. **Complex Handlers Functional**
- Waiver wire analysis (most complex handler) working
- Lineup optimization functioning correctly
- Multi-level roster data retrieval operational
5. **No Regressions**
- All previously working features still functional
- No breaking changes from refactoring
- Backward compatibility maintained
### 📊 Performance Analysis
**Fast Operations** (< 0.5s):
- Admin operations (cache, status)
- League discovery (cached)
- Basic roster retrieval
**Medium Operations** (0.5-2s):
- League info retrieval
- Standings
- Player searches
- Waiver wire analysis
**Slow Operations** (> 2s):
- Full roster with external data (multi-source)
- Reddit sentiment analysis (external API)
All performance metrics are within expected ranges. No performance regressions detected.
## Test Infrastructure
### Test Script Features
The `test_live_api.py` script provides:
1. **Comprehensive Coverage**
- Tests all 18 MCP tools
- Covers all 7 handler domains
- Tests different configuration levels (basic/standard/full roster)
2. **Intelligent Test Flow**
- Auto-discovers leagues for context
- Sets up test data dynamically
- Handles missing data gracefully
3. **User-Friendly Output**
- Color-coded results (green=pass, red=fail, yellow=warning)
- Progress indicators
- Timing metrics
- Summary statistics
4. **Safety Features**
- Rate limit checking before testing
- Delays between tests (0.2s)
- Graceful error handling
- Continues testing even if one test fails
5. **Automated Documentation**
- Generates `LIVE_API_TEST_RESULTS.md`
- Includes performance metrics
- Provides recommendations
- Timestamps and environment info
## Test Scenarios Covered
### 1. Admin Operations
- API status checking
- Cache clearing
- Token refreshing
### 2. League Discovery & Management
- Multi-league discovery
- League info retrieval
- Standings
- Team listings
### 3. Roster Management
- Basic roster (quick info)
- Standard roster (with projections)
- Full roster (enhanced with external data)
### 4. Matchup Analysis
- Current week matchup retrieval
- Team comparisons
- Lineup optimization with strategies
### 5. Player Research
- Position-filtered player searches
- Free agent queries
- Waiver wire analysis
### 6. Draft Tools
- Draft results/recap
- Pre-draft rankings
- Live draft recommendations
- Draft state analysis
### 7. Analytics
- Multi-player Reddit sentiment analysis
- Social media engagement metrics
## Files Generated
### 1. `test_live_api.py` (650 lines)
Comprehensive test script with:
- Test orchestration
- Result tracking
- Performance metrics
- Document generation
- Color-coded output
### 2. `LIVE_API_TEST_RESULTS.md`
Detailed results document with:
- Executive summary
- Per-category results
- Performance metrics
- Recommendations
### 3. Updated `PHASE_2B_REFACTOR_SUMMARY.md`
Added live API testing section with:
- Test results
- Performance data
- Verification status
## Assessment
### ✅ EXCELLENT - Production Ready
**Criteria Met**:
- ✅ 100% test pass rate (22/22)
- ✅ All handler domains functional
- ✅ No regressions detected
- ✅ Performance within expectations
- ✅ External integrations working
- ✅ Dependency injection verified
**Recommendation**: **READY TO MERGE** to main branch
## Usage
### Running the Tests
```bash
# Run full test suite
python test_live_api.py
# Verify environment first
python -c "from dotenv import load_dotenv; import os; load_dotenv(); \
print('✓ Ready' if all([os.getenv(v) for v in ['YAHOO_CONSUMER_KEY', \
'YAHOO_ACCESS_TOKEN', 'YAHOO_GUID']]) else '✗ Missing env vars')"
# Refresh token if needed
python -c "from fantasy_football_multi_league import call_tool; \
import asyncio; asyncio.run(call_tool('ff_refresh_token', {}))"
```
### Test Output
The test script provides real-time feedback:
- Color-coded pass/fail indicators
- Timing for each test
- Category summaries
- Overall statistics
- Auto-generated results document
### Exit Codes
- `0` - All tests passed (≥90% pass rate)
- `1` - Too many failures (<80% pass rate)
## Lessons Learned
### What Worked Well
1. **Comprehensive Test Coverage** - Testing all handlers caught edge cases
2. **Live API Testing** - Verified real-world functionality beyond unit tests
3. **Automated Documentation** - Results doc provides clear record
4. **Safety Features** - Rate limiting and delays prevent API abuse
5. **Graceful Degradation** - Tests continue even if some fail
### Challenges Addressed
1. **Response Format Variations** - Fixed parser to handle both status-based and direct responses
2. **Token Expiration** - Added automatic token refresh at start
3. **Context Setup** - Auto-discovery of leagues/teams for dynamic testing
4. **External API Delays** - Reddit API slower but functional
### Best Practices Established
1. Always refresh token before live testing
2. Use delays between tests to respect rate limits
3. Test with real data, not just mocks
4. Generate documentation automatically
5. Provide clear visual feedback during testing
## Next Steps
### Immediate Actions
1. ✅ **Merge to main** - All verification complete
2. ✅ **Update documentation** - Live test results recorded
3. ✅ **Archive test artifacts** - Save test script and results
### Future Enhancements
1. **CI/CD Integration** - Add live API tests to deployment pipeline
2. **Scheduled Testing** - Run periodic health checks
3. **Performance Monitoring** - Track response times over time
4. **Extended Coverage** - Add edge case testing (injuries, trades, etc.)
## Conclusion
The live API testing successfully verified that Phase 2b refactoring achieved its goals without introducing regressions. All 22 tests passed, demonstrating that:
- ✅ Handler extraction was successful
- ✅ Dependency injection pattern works correctly
- ✅ All API integrations remain functional
- ✅ Performance is within acceptable ranges
- ✅ Code is production-ready
The refactoring reduced the main file by 46% (999 lines) while maintaining 100% functionality. The codebase is now more maintainable, testable, and scalable.
**Status**: ✅ **VERIFIED & APPROVED FOR PRODUCTION**
---
*Generated on October 2, 2025*
*Test Script: `test_live_api.py`*
*Results: `LIVE_API_TEST_RESULTS.md`*
```
--------------------------------------------------------------------------------
/src/yahoo_api_utils.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Yahoo API utilities for rate limiting and caching
"""
import asyncio
import time
import hashlib
import json
from typing import Any, Dict, Optional, Callable
from functools import wraps
from collections import deque
from datetime import datetime, timedelta
class RateLimiter:
"""Rate limiter for Yahoo API calls (1000 requests per hour)."""
def __init__(self, max_requests: int = 900, window_seconds: int = 3600):
"""
Initialize rate limiter.
Using 900 instead of 1000 to have safety margin.
Args:
max_requests: Maximum requests allowed in the window
window_seconds: Time window in seconds (3600 = 1 hour)
"""
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self._lock = asyncio.Lock()
async def acquire(self):
"""Wait if necessary to respect rate limits."""
async with self._lock:
now = time.time()
# Remove old requests outside the window
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
# If at limit, calculate wait time
if len(self.requests) >= self.max_requests:
oldest_request = self.requests[0]
wait_time = (oldest_request + self.window_seconds) - now
if wait_time > 0:
print(f"Rate limit reached. Waiting {wait_time:.1f} seconds...")
await asyncio.sleep(wait_time)
# After waiting, clean up old requests again
now = time.time()
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
# Record this request
self.requests.append(now)
def get_status(self) -> Dict[str, Any]:
"""Get current rate limiter status."""
now = time.time()
# Clean old requests
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
requests_in_window = len(self.requests)
remaining = self.max_requests - requests_in_window
# Calculate reset time (when oldest request expires)
reset_time = None
if self.requests:
reset_time = self.requests[0] + self.window_seconds
reset_in_seconds = max(0, reset_time - now)
else:
reset_in_seconds = 0
return {
"requests_used": requests_in_window,
"requests_remaining": remaining,
"max_requests": self.max_requests,
"reset_in_seconds": round(reset_in_seconds),
"reset_time": datetime.fromtimestamp(reset_time).isoformat() if reset_time else None
}
class ResponseCache:
"""Simple TTL-based cache for API responses."""
def __init__(self):
self.cache: Dict[str, tuple[Any, float]] = {}
self._lock = asyncio.Lock()
# Default TTLs for different endpoint types (in seconds)
self.default_ttls = {
"leagues": 3600, # 1 hour - leagues don't change often
"teams": 1800, # 30 minutes - team info fairly static
"standings": 300, # 5 minutes - standings update after games
"roster": 300, # 5 minutes - roster changes matter
"matchup": 60, # 1 minute - live scoring during games
"players": 600, # 10 minutes - free agents change slowly
"draft": 86400, # 24 hours - draft results are static
"waiver": 300, # 5 minutes - waiver wire is dynamic
"user": 3600, # 1 hour - user info rarely changes
}
def _get_cache_key(self, endpoint: str) -> str:
"""Generate cache key from endpoint."""
return hashlib.md5(endpoint.encode()).hexdigest()
def _get_ttl_for_endpoint(self, endpoint: str) -> int:
"""Determine TTL based on endpoint type."""
# Check endpoint patterns to determine type
if "leagues" in endpoint or "games" in endpoint:
return self.default_ttls["leagues"]
elif "standings" in endpoint:
return self.default_ttls["standings"]
elif "roster" in endpoint:
return self.default_ttls["roster"]
elif "matchup" in endpoint or "scoreboard" in endpoint:
return self.default_ttls["matchup"]
elif "players" in endpoint and "status=A" in endpoint:
return self.default_ttls["players"]
elif "draft" in endpoint:
return self.default_ttls["draft"]
elif "teams" in endpoint:
return self.default_ttls["teams"]
elif "users" in endpoint:
return self.default_ttls["user"]
else:
return 300 # Default 5 minutes
async def get(self, endpoint: str) -> Optional[Any]:
"""Get cached response if valid."""
async with self._lock:
cache_key = self._get_cache_key(endpoint)
if cache_key in self.cache:
data, timestamp = self.cache[cache_key]
ttl = self._get_ttl_for_endpoint(endpoint)
if time.time() - timestamp < ttl:
age = time.time() - timestamp
return data
else:
# Expired, remove from cache
del self.cache[cache_key]
return None
async def set(self, endpoint: str, data: Any):
"""Store response in cache."""
async with self._lock:
cache_key = self._get_cache_key(endpoint)
self.cache[cache_key] = (data, time.time())
async def clear(self, pattern: Optional[str] = None):
"""Clear cache entries matching pattern or all if no pattern."""
async with self._lock:
if pattern:
keys_to_delete = [
key for key in self.cache.keys()
if pattern in key
]
for key in keys_to_delete:
del self.cache[key]
else:
self.cache.clear()
def get_stats(self) -> Dict[str, Any]:
"""Get cache statistics."""
now = time.time()
total_entries = len(self.cache)
expired_count = 0
total_size = 0
for endpoint_hash, (data, timestamp) in self.cache.items():
# Estimate size (rough)
total_size += len(json.dumps(data, default=str))
# Find endpoint type from hash (approximate)
for endpoint_type in self.default_ttls:
ttl = self.default_ttls[endpoint_type]
if now - timestamp >= ttl:
expired_count += 1
break
return {
"total_entries": total_entries,
"expired_entries": expired_count,
"active_entries": total_entries - expired_count,
"cache_size_bytes": total_size,
"cache_size_mb": round(total_size / (1024 * 1024), 2)
}
# Global instances
rate_limiter = RateLimiter()
response_cache = ResponseCache()
def with_rate_limit(func: Callable) -> Callable:
"""Decorator to add rate limiting to async functions."""
@wraps(func)
async def wrapper(*args, **kwargs):
await rate_limiter.acquire()
return await func(*args, **kwargs)
return wrapper
def with_cache(ttl_seconds: Optional[int] = None) -> Callable:
"""
Decorator to add caching to async functions.
Args:
ttl_seconds: Override default TTL for this function
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(endpoint: str, *args, **kwargs):
# Try to get from cache first
cached_response = await response_cache.get(endpoint)
if cached_response is not None:
return cached_response
# Not in cache, make the actual call
result = await func(endpoint, *args, **kwargs)
# Store in cache
if result: # Only cache successful responses
await response_cache.set(endpoint, result)
return result
return wrapper
return decorator
```
--------------------------------------------------------------------------------
/position_normalizer.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Position Normalizer for Fantasy Football
Normalizes player scores across positions for fair FLEX comparisons.
Uses historical averages and standard deviations to create z-scores.
"""
from typing import Dict, Tuple
import numpy as np
class PositionNormalizer:
"""Normalizes fantasy scores across positions for fair comparison."""
def __init__(self):
# 2023 PPR scoring averages by position (based on starters)
# These are typical weekly averages for fantasy-relevant players
self.position_stats = {
"QB": {
"mean": 18.0,
"std": 6.0,
"starter_threshold": 15.0, # QB12 level
"elite_threshold": 22.0, # QB3 level
},
"RB": {
"mean": 11.0,
"std": 5.0,
"starter_threshold": 8.0, # RB24 level
"elite_threshold": 15.0, # RB5 level
},
"WR": {
"mean": 10.0,
"std": 4.5,
"starter_threshold": 7.0, # WR36 level
"elite_threshold": 14.0, # WR5 level
},
"TE": {
"mean": 7.0,
"std": 3.5,
"starter_threshold": 5.0, # TE12 level
"elite_threshold": 10.0, # TE3 level
},
"K": {"mean": 8.0, "std": 3.0, "starter_threshold": 6.0, "elite_threshold": 10.0},
"DEF": {"mean": 8.0, "std": 4.0, "starter_threshold": 6.0, "elite_threshold": 12.0},
}
def normalize_projection(self, projection: float, position: str) -> float:
"""
Convert raw projection to normalized score (z-score).
A normalized score of:
- 0 = average for that position
- 1 = 1 standard deviation above average (good)
- 2 = 2 standard deviations above average (elite)
- -1 = 1 standard deviation below average (poor)
Args:
projection: Raw fantasy point projection
position: Player position
Returns:
Normalized score (z-score)
"""
if position not in self.position_stats:
return 0.0
stats = self.position_stats[position]
z_score = (projection - stats["mean"]) / stats["std"]
return z_score
def get_flex_value(self, projection: float, position: str) -> float:
"""
Calculate FLEX value score that accounts for position scarcity.
This creates a fair comparison across positions by considering:
1. How much above replacement level the player is
2. Position scarcity factor
Args:
projection: Raw fantasy point projection
position: Player position
Returns:
FLEX value score (higher = better FLEX play)
"""
if position not in self.position_stats:
return projection # Fallback to raw projection
stats = self.position_stats[position]
# Value Over Replacement (VOR)
# Replacement level is roughly the starter threshold
replacement_level = stats["starter_threshold"]
vor = projection - replacement_level
# Position scarcity multiplier
# TEs have fewer elite options, so good TEs get a small boost
scarcity_factors = {
"RB": 1.0, # Baseline
"WR": 0.95, # Slightly more WRs available
"TE": 1.05, # Slight boost for scarcity, but not too much
}
scarcity = scarcity_factors.get(position, 1.0)
# FLEX value = VOR * scarcity + baseline projection weight
# We still weight the raw projection heavily to avoid TE trap
flex_value = (vor * scarcity * 0.3) + (projection * 0.7)
return flex_value
def get_percentile_rank(self, projection: float, position: str) -> float:
"""
Get percentile rank for a projection within position.
Args:
projection: Raw fantasy point projection
position: Player position
Returns:
Percentile (0-100, where 100 = best)
"""
if position not in self.position_stats:
return 50.0
stats = self.position_stats[position]
z_score = self.normalize_projection(projection, position)
# Convert z-score to percentile using normal CDF approximation
# This is a simplified version - could use scipy.stats.norm.cdf
percentile = 50 + (z_score * 33.3) # Rough approximation
percentile = max(0, min(100, percentile))
return percentile
def is_starter_worthy(self, projection: float, position: str) -> bool:
"""
Determine if a player is starter-worthy at their position.
Args:
projection: Raw fantasy point projection
position: Player position
Returns:
True if starter-worthy, False otherwise
"""
if position not in self.position_stats:
return projection >= 7.0 # Generic threshold
return projection >= self.position_stats[position]["starter_threshold"]
def is_elite(self, projection: float, position: str) -> bool:
"""
Determine if a player is elite at their position.
Args:
projection: Raw fantasy point projection
position: Player position
Returns:
True if elite, False otherwise
"""
if position not in self.position_stats:
return projection >= 15.0 # Generic threshold
return projection >= self.position_stats[position]["elite_threshold"]
def compare_for_flex(self, player_a: Tuple[float, str], player_b: Tuple[float, str]) -> str:
"""
Compare two players for FLEX spot using normalized scoring.
Args:
player_a: (projection, position) for player A
player_b: (projection, position) for player B
Returns:
"A" if player A is better, "B" if player B is better
"""
proj_a, pos_a = player_a
proj_b, pos_b = player_b
# Get FLEX values
flex_a = self.get_flex_value(proj_a, pos_a)
flex_b = self.get_flex_value(proj_b, pos_b)
# If FLEX values are very close (within 0.3 points),
# use raw projection as tiebreaker
if abs(flex_a - flex_b) < 0.3:
return "A" if proj_a >= proj_b else "B"
return "A" if flex_a > flex_b else "B"
def explain_comparison(
self, player_a: Tuple[float, str, str], player_b: Tuple[float, str, str]
) -> str:
"""
Explain why one player is better for FLEX than another.
Args:
player_a: (projection, position, name) for player A
player_b: (projection, position, name) for player B
Returns:
Explanation string
"""
proj_a, pos_a, name_a = player_a
proj_b, pos_b, name_b = player_b
# Calculate various metrics
z_a = self.normalize_projection(proj_a, pos_a)
z_b = self.normalize_projection(proj_b, pos_b)
flex_a = self.get_flex_value(proj_a, pos_a)
flex_b = self.get_flex_value(proj_b, pos_b)
pct_a = self.get_percentile_rank(proj_a, pos_a)
pct_b = self.get_percentile_rank(proj_b, pos_b)
winner = name_a if flex_a > flex_b else name_b
explanation = f"""
FLEX Comparison: {name_a} ({pos_a}) vs {name_b} ({pos_b})
{name_a} ({pos_a}):
- Projection: {proj_a:.1f} points
- Position Percentile: {pct_a:.0f}th
- Normalized Score: {z_a:+.2f} ({"above" if z_a > 0 else "below"} average)
- FLEX Value: {flex_a:.1f}
{name_b} ({pos_b}):
- Projection: {proj_b:.1f} points
- Position Percentile: {pct_b:.0f}th
- Normalized Score: {z_b:+.2f} ({"above" if z_b > 0 else "below"} average)
- FLEX Value: {flex_b:.1f}
Winner: {winner}
Reason: {"Higher FLEX value accounting for position and scarcity" if abs(flex_a - flex_b) >= 0.3 else "Higher raw projection (tiebreaker)"}
"""
return explanation
# Global instance
position_normalizer = PositionNormalizer()
# Example usage
if __name__ == "__main__":
normalizer = PositionNormalizer()
# Example: 9-point TE vs 10-point RB
print("Example 1: 9-point TE vs 10-point RB")
print("-" * 40)
te_proj = 9.0
rb_proj = 10.0
winner = normalizer.compare_for_flex((te_proj, "TE"), (rb_proj, "RB"))
print(f"Winner: {'TE' if winner == 'A' else 'RB'}")
explanation = normalizer.explain_comparison(
(te_proj, "TE", "George Kittle"), (rb_proj, "RB", "Tony Pollard")
)
print(explanation)
# Example 2: 8-point WR vs 8-point RB (true tie)
print("\nExample 2: 8-point WR vs 8-point RB")
print("-" * 40)
wr_proj = 8.0
rb_proj = 8.0
winner = normalizer.compare_for_flex((wr_proj, "WR"), (rb_proj, "RB"))
print(f"Winner: {'WR' if winner == 'A' else 'RB'}")
explanation = normalizer.explain_comparison(
(wr_proj, "WR", "Chris Olave"), (rb_proj, "RB", "Dameon Pierce")
)
print(explanation)
```
--------------------------------------------------------------------------------
/docs/TEST_SUITE_SUMMARY.md:
--------------------------------------------------------------------------------
```markdown
# Test Suite Implementation Summary
## Project: Fantasy Football MCP Server - Comprehensive Test Suite
**Date**: January 2025
**Objective**: Build comprehensive pytest test suite for stability and future refactoring confidence
---
## 🎯 What Was Accomplished
### Test Suite Created
Successfully created a comprehensive test suite with **70 passing tests**:
- **62 Unit Tests**: Testing individual modules in isolation
- **8 Integration Tests**: Testing complete workflows end-to-end
### Test Files Created
```
tests/
├── conftest.py # Shared fixtures (300+ lines)
├── README.md # Complete testing documentation
├── unit/
│ ├── __init__.py
│ ├── test_api_client.py # 12 tests - Yahoo API client
│ ├── test_handlers.py # 9 tests - MCP tool handlers
│ ├── test_lineup_optimizer.py # 29 tests - Optimization logic
│ └── test_parsers.py # 12 tests - Yahoo API parsing
└── integration/
├── __init__.py
└── test_mcp_tools.py # 8 tests - End-to-end flows
```
### Coverage Achieved
| Module | Lines | Coverage | Status |
|--------|-------|----------|---------|
| **src/api/yahoo_client.py** | 64 | **97%** | ✅ Excellent |
| **src/handlers/admin_handlers.py** | 12 | **100%** | ✅ Perfect |
| **src/handlers/league_handlers.py** | 64 | **86%** | ✅ Very Good |
| **src/parsers/yahoo_parsers.py** | 125 | **86%** | ✅ Very Good |
| **lineup_optimizer.py** | 312 | **51%** | ✅ Good (complex module) |
**Critical paths have 80%+ coverage** - exceeding the target!
---
## 📝 Test Categories
### Unit Tests (62 tests)
#### API Client Tests (12 tests)
- ✅ Token management (get/set)
- ✅ Yahoo API calls with caching
- ✅ Rate limiting integration
- ✅ Automatic token refresh on 401
- ✅ Error handling (500 errors)
- ✅ Cache disabled mode
#### Handler Tests (9 tests)
- ✅ Admin handlers (refresh_token, get_api_status, clear_cache)
- ✅ League handlers (get_leagues, get_league_info, get_standings, get_teams)
- ✅ Error handling for missing parameters
- ✅ Response parsing and formatting
#### Lineup Optimizer Tests (29 tests)
- ✅ Utility functions (coerce_float, coerce_int, normalize_position)
- ✅ Match analytics tracking
- ✅ Player dataclass and validation
- ✅ Roster parsing from different formats
- ✅ Position normalization edge cases
- ✅ Invalid/malformed data handling
#### Parser Tests (12 tests)
- ✅ Team roster parsing
- ✅ Free agent/waiver wire parsing
- ✅ Malformed response handling
- ✅ Selected position precedence
- ✅ Nested team structure extraction
- ✅ Ownership and injury data
### Integration Tests (8 tests)
#### MCP Tool Flows (8 tests)
- ✅ Complete league workflow (discover → info → standings)
- ✅ Roster parsing to lineup optimizer pipeline
- ✅ Token refresh and status check workflow
- ✅ Yahoo API → Parser → Optimizer transformation
- ✅ Error recovery in multi-stage pipelines
- ✅ Caching behavior (hits and misses)
---
## 🔧 Technical Implementation
### Test Infrastructure
**Fixtures Created** (`conftest.py`):
- `mock_env_vars`: Yahoo API credential mocks
- `mock_yahoo_league_response`: Sample league API data
- `mock_yahoo_roster_response`: Sample roster API data
- `mock_yahoo_free_agents_response`: Sample free agent data
- `mock_yahoo_standings_response`: Sample standings data
- `mock_rate_limiter`: Rate limiter mock
- `mock_response_cache`: Response cache mock
- `sample_roster_data`: Parsed roster samples
- `sample_sleeper_rankings`: Sleeper API samples
### Async Testing
Successfully implemented async testing with proper mocking:
```python
# Proper async context manager mocking for aiohttp
class MockSession:
def get(self, *args, **kwargs):
class Context:
async def __aenter__(self):
return MockResponse()
async def __aexit__(self, *args):
return None
return Context()
```
### Configuration
Added to `requirements.txt`:
```
pytest==8.4.2
pytest-asyncio==1.2.0
pytest-mock==3.15.1
pytest-cov==6.0.0 # NEW: Code coverage reporting
```
Configured in `pyproject.toml`:
```toml
[tool.pytest.ini_options]
minversion = "6.0"
testpaths = ["tests"]
pythonpath = ["."]
asyncio_mode = "auto"
markers = [
"slow: marks tests as slow",
"integration: marks tests as integration tests",
"unit: marks tests as unit tests",
]
```
---
## 🎓 Key Learnings & Solutions
### Challenge 1: Async Context Manager Mocking
**Problem**: AsyncMock doesn't work correctly with aiohttp's async context managers
**Solution**: Create proper async context manager classes:
```python
class MockGetContext:
async def __aenter__(self):
return MockResponse()
async def __aexit__(self, *args):
return None
```
### Challenge 2: Position Normalization Logic
**Problem**: Tests assumed position transformation (D/ST → DEF) but code only uppercases
**Solution**: Updated tests to match actual behavior - `_normalize_position` uppercases only
### Challenge 3: Integration Test Scope
**Problem**: Balancing thorough testing with fast test execution
**Solution**: Created focused integration tests for critical paths only
---
## ✅ Benefits Achieved
### 1. **Confidence for Future Refactoring**
- Can now safely extract remaining handlers (Phase 2b)
- Tests will catch breaking changes immediately
- Refactoring can proceed incrementally with test validation
### 2. **Code Quality Assurance**
- Critical modules have 80%+ coverage
- Edge cases are tested (empty responses, malformed data)
- Error handling is validated
### 3. **Documentation**
- Tests serve as usage examples
- Clear test names document expected behavior
- Test README provides onboarding for new developers
### 4. **Regression Prevention**
- Any new changes must pass all 70 tests
- Future bugs can be captured as test cases
- CI/CD ready for automated testing
---
## 📊 Test Execution Performance
```bash
$ pytest tests/unit/ tests/integration/ -v
================================ 70 passed in 4.09s =================================
```
**Fast execution**: < 5 seconds for full suite
**Stable**: 100% pass rate
**Maintainable**: Well-organized, documented, and uses fixtures
---
## 🚀 Running the Tests
### Quick Start
```bash
# Install dependencies
pip install -r requirements.txt
# Run all tests
pytest tests/ -v
# Run with coverage
pytest tests/ --cov=src --cov=lineup_optimizer --cov-report=term-missing
```
### Specific Test Types
```bash
# Unit tests only
pytest tests/unit/ -v
# Integration tests only
pytest tests/integration/ -v
# Tests matching pattern
pytest tests/ -k "test_yahoo" -v
```
### Coverage Report
```bash
# Generate HTML coverage report
pytest tests/ --cov=src --cov=lineup_optimizer --cov-report=html
# View in browser
open htmlcov/index.html
```
---
## 📈 Next Steps (Optional)
### Phase 2b: Extract Remaining Handlers
Now that we have comprehensive tests, Phase 2b refactoring is **much safer**:
1. ✅ **Tests provide safety net**: Any breaking changes will be caught
2. ✅ **Can refactor incrementally**: One handler at a time
3. ✅ **Quick feedback**: Run tests after each change
### Future Test Enhancements
- [ ] Add tests for draft evaluation algorithms
- [ ] Add tests for matchup analyzer
- [ ] Add tests for position normalizer
- [ ] Add tests for Sleeper API integration
- [ ] Add performance/benchmark tests
- [ ] Increase coverage for complex handlers (70% → 85%+)
---
## 🎉 Success Metrics
✅ **70 tests created** (62 unit + 8 integration)
✅ **100% pass rate**
✅ **80%+ coverage** on critical modules
✅ **< 5 second** test execution
✅ **Comprehensive documentation** (tests/README.md)
✅ **Production-ready** test infrastructure
---
## 📚 Files Modified/Created
### New Files (7)
- `tests/conftest.py` (300+ lines)
- `tests/README.md` (400+ lines)
- `tests/unit/__init__.py`
- `tests/unit/test_api_client.py` (270+ lines)
- `tests/unit/test_handlers.py` (130+ lines)
- `tests/unit/test_lineup_optimizer.py` (350+ lines)
- `tests/unit/test_parsers.py` (240+ lines)
- `tests/integration/__init__.py`
- `tests/integration/test_mcp_tools.py` (280+ lines)
- `TEST_SUITE_SUMMARY.md` (this file)
### Modified Files (1)
- `requirements.txt` (added pytest-cov)
---
## 💡 Recommendations
### For Immediate Use
1. **Run tests before commits**:
```bash
pytest tests/ -v
```
2. **Check coverage for new code**:
```bash
pytest tests/ --cov=src/new_module --cov-report=term-missing
```
3. **Keep tests passing**: Don't merge code that breaks tests
### For Future Development
1. **Write tests first** for new features (TDD)
2. **Add test cases** when bugs are discovered
3. **Maintain 80%+ coverage** for critical paths
4. **Update tests** when refactoring code
5. **Document test fixtures** for complex scenarios
---
## 🙏 Conclusion
Successfully created a **comprehensive, maintainable, and production-ready test suite** for the Fantasy Football MCP Server. The test infrastructure provides:
- ✅ Confidence for future refactoring
- ✅ Regression prevention
- ✅ Code quality assurance
- ✅ Developer documentation
- ✅ CI/CD readiness
**The codebase is now significantly more robust and ready for continued development!**
---
**Questions or Issues?**
See `tests/README.md` for detailed testing documentation and troubleshooting guide.
```
--------------------------------------------------------------------------------
/src/api/yahoo_utils.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Yahoo API utilities for rate limiting and caching
"""
import asyncio
import time
import hashlib
import json
from typing import Any, Dict, Optional, Callable
from functools import wraps
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timedelta
class RateLimiter:
"""Rate limiter for Yahoo API calls (1000 requests per hour)."""
def __init__(self, max_requests: int = 900, window_seconds: int = 3600):
"""
Initialize rate limiter.
Using 900 instead of 1000 to have safety margin.
Args:
max_requests: Maximum requests allowed in the window
window_seconds: Time window in seconds (3600 = 1 hour)
"""
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self._lock = asyncio.Lock()
async def acquire(self):
"""Wait if necessary to respect rate limits."""
async with self._lock:
now = time.time()
# Remove old requests outside the window
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
# If at limit, calculate wait time
if len(self.requests) >= self.max_requests:
oldest_request = self.requests[0]
wait_time = (oldest_request + self.window_seconds) - now
if wait_time > 0:
print(f"Rate limit reached. Waiting {wait_time:.1f} seconds...")
await asyncio.sleep(wait_time)
# After waiting, clean up old requests again
now = time.time()
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
# Record this request
self.requests.append(now)
def get_status(self) -> Dict[str, Any]:
"""Get current rate limiter status."""
now = time.time()
# Clean old requests
while self.requests and self.requests[0] <= now - self.window_seconds:
self.requests.popleft()
requests_in_window = len(self.requests)
remaining = self.max_requests - requests_in_window
# Calculate reset time (when oldest request expires)
reset_time = None
if self.requests:
reset_time = self.requests[0] + self.window_seconds
reset_in_seconds = max(0, reset_time - now)
else:
reset_in_seconds = 0
return {
"requests_used": requests_in_window,
"requests_remaining": remaining,
"max_requests": self.max_requests,
"reset_in_seconds": round(reset_in_seconds),
"reset_time": datetime.fromtimestamp(reset_time).isoformat() if reset_time else None,
}
@dataclass
class CacheEntry:
data: Any
timestamp: float
endpoint: str
ttl: int
@property
def age(self) -> float:
return time.time() - self.timestamp
class ResponseCache:
"""Simple TTL-based cache for API responses."""
def __init__(self):
self.cache: Dict[str, CacheEntry] = {}
self._lock = asyncio.Lock()
# Default TTLs for different endpoint types (in seconds)
self.default_ttls = {
"leagues": 3600, # 1 hour - leagues don't change often
"teams": 1800, # 30 minutes - team info fairly static
"standings": 300, # 5 minutes - standings update after games
"roster": 300, # 5 minutes - roster changes matter
"matchup": 60, # 1 minute - live scoring during games
"players": 600, # 10 minutes - free agents change slowly
"draft": 86400, # 24 hours - draft results are static
"waiver": 300, # 5 minutes - waiver wire is dynamic
"user": 3600, # 1 hour - user info rarely changes
}
def _get_cache_key(self, endpoint: str) -> str:
"""Generate cache key from endpoint."""
return hashlib.md5(endpoint.encode()).hexdigest()
def _get_ttl_for_endpoint(self, endpoint: str) -> int:
"""Determine TTL based on endpoint type."""
if "leagues" in endpoint or "games" in endpoint:
return self.default_ttls["leagues"]
if "standings" in endpoint:
return self.default_ttls["standings"]
if "roster" in endpoint:
return self.default_ttls["roster"]
if "matchup" in endpoint or "scoreboard" in endpoint:
return self.default_ttls["matchup"]
if "players" in endpoint and "status=A" in endpoint:
return self.default_ttls["players"]
if "draft" in endpoint:
return self.default_ttls["draft"]
if "teams" in endpoint:
return self.default_ttls["teams"]
if "users" in endpoint:
return self.default_ttls["user"]
return 300 # Default 5 minutes
async def get(self, endpoint: str) -> Optional[Any]:
"""Get cached response if valid."""
cache_key = self._get_cache_key(endpoint)
async with self._lock:
entry = self.cache.get(cache_key)
if entry is None:
return None
if not isinstance(entry, CacheEntry):
data, timestamp = entry # Backwards compatibility for old entries
ttl = self._get_ttl_for_endpoint(endpoint)
if time.time() - timestamp < ttl:
return data
del self.cache[cache_key]
return None
if entry.age < entry.ttl:
return entry.data
del self.cache[cache_key]
return None
async def set(self, endpoint: str, data: Any, ttl: Optional[int] = None):
"""Store response in cache."""
cache_key = self._get_cache_key(endpoint)
ttl_value = ttl if ttl is not None else self._get_ttl_for_endpoint(endpoint)
async with self._lock:
self.cache[cache_key] = CacheEntry(
data=data,
timestamp=time.time(),
endpoint=endpoint,
ttl=ttl_value,
)
async def clear(self, pattern: Optional[str] = None):
"""Clear cache entries matching pattern or all if no pattern."""
async with self._lock:
if pattern:
keys_to_delete = [
key
for key, entry in self.cache.items()
if pattern in key
or (isinstance(entry, CacheEntry) and pattern in entry.endpoint)
]
for key in keys_to_delete:
del self.cache[key]
else:
self.cache.clear()
def get_stats(self) -> Dict[str, Any]:
"""Get cache statistics."""
now = time.time()
entries = list(self.cache.values())
total_entries = len(entries)
expired_count = 0
total_size = 0
sample_endpoints = []
oldest_age = 0.0
for entry in entries:
if isinstance(entry, CacheEntry):
ttl = entry.ttl
endpoint = entry.endpoint
timestamp = entry.timestamp
data = entry.data
sample_endpoints.append(endpoint)
else:
data, timestamp = entry
endpoint = "unknown"
ttl = self._get_ttl_for_endpoint(endpoint)
if now - timestamp >= ttl:
expired_count += 1
age = now - timestamp
if age > oldest_age:
oldest_age = age
try:
total_size += len(json.dumps(data, default=str))
except TypeError:
pass
return {
"total_entries": total_entries,
"expired_entries": expired_count,
"active_entries": total_entries - expired_count,
"cache_size_bytes_est": total_size,
"cache_size_mb_est": round(total_size / (1024 * 1024), 2),
"oldest_entry_age_seconds": round(oldest_age, 1),
"sample_endpoints": sample_endpoints[:5],
}
# Global instances
rate_limiter = RateLimiter()
response_cache = ResponseCache()
def with_rate_limit(func: Callable) -> Callable:
"""Decorator to add rate limiting to async functions."""
@wraps(func)
async def wrapper(*args, **kwargs):
await rate_limiter.acquire()
return await func(*args, **kwargs)
return wrapper
def with_cache(ttl_seconds: Optional[int] = None) -> Callable:
"""
Decorator to add caching to async functions.
Args:
ttl_seconds: Override default TTL for this function
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(endpoint: str, *args, **kwargs):
# Try to get from cache first
cached_response = await response_cache.get(endpoint)
if cached_response is not None:
return cached_response
# Not in cache, make the actual call
result = await func(endpoint, *args, **kwargs)
# Store in cache
if result: # Only cache successful responses
await response_cache.set(endpoint, result, ttl=ttl_seconds)
return result
return wrapper
return decorator
```