10 Commits

Author SHA1 Message Date
matthias 99fd37fc12 test: implement unit testing suite with pytest and add pre-push verification hook
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m21s
2026-05-13 00:12:32 +02:00
matthias 8e9e4c01d4 style: refactor codebase to adhere to PEP 8 formatting standards throughout all source files 2026-05-12 23:55:58 +02:00
matthias bcc11cb07e refactor: remove unused ContentBlock import from tool modules
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m26s
2026-05-12 23:38:19 +02:00
matthias 40b5d004b1 refactor: enhance tool outputs by returning formatted markdown and JSON resources for structured display
CI/CD Pipeline / Lint & Check (push) Failing after 10s
CI/CD Pipeline / Build & Push Docker Image (push) Has been skipped
2026-05-12 23:36:37 +02:00
matthias 7c089d90c5 refactor: remove return type annotations from activity and athlete tools
CI/CD Pipeline / Lint & Check (push) Failing after 9s
CI/CD Pipeline / Build & Push Docker Image (push) Has been skipped
2026-05-12 23:28:47 +02:00
matthias c69e362635 refactor: migrate tool outputs to use EmbeddedResource with typed JSON for assistant-facing data
CI/CD Pipeline / Lint & Check (push) Successful in 9s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m20s
2026-05-12 23:23:23 +02:00
matthias 3805ca3274 feat: standardize on ISO 8601 for dates, add utility functions, and document design decisions.
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m21s
2026-05-12 23:09:00 +02:00
matthias c56f7ad7b4 refactor: remove interactive OAuth tool and update Docker/README configurations
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m19s
2026-05-10 11:44:39 +02:00
matthias 578c4b292a feat: enhance OAuth flow with synchronous token exchange and automatic .env file updates
CI/CD Pipeline / Lint & Check (push) Failing after 9s
CI/CD Pipeline / Build & Push Docker Image (push) Has been skipped
2026-05-10 11:15:00 +02:00
matthias fafda14fe9 refactor: improve Docker image pull string construction in Gitea CI workflow using jq arguments
CI/CD Pipeline / Lint & Check (push) Successful in 10s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m21s
2026-05-09 15:04:20 +02:00
27 changed files with 1481 additions and 447 deletions
+2 -3
View File
@@ -103,9 +103,8 @@ jobs:
# Check if Docker Image section already exists # Check if Docker Image section already exists
if [[ "$OLD_BODY" != *"## 🐳 Docker Image"* ]]; then if [[ "$OLD_BODY" != *"## 🐳 Docker Image"* ]]; then
NEW_BODY="${OLD_BODY}\n\n## 🐳 Docker Image\n\`\`\`bash\ndocker pull ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${TAG_NAME}\n\`\`\`" jq -n --arg old "$OLD_BODY" --arg img "${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${TAG_NAME}" \
'{body: ($old + "\n\n## 🐳 Docker Image\n```bash\ndocker pull " + $img + "\n```")}' | \
jq -n --arg body "$NEW_BODY" '{body: $body}' | \
curl -s -X PATCH \ curl -s -X PATCH \
-H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \ -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
-H "Content-Type: application/json" \ -H "Content-Type: application/json" \
+5 -3
View File
@@ -27,8 +27,10 @@ RUN uv sync --frozen --no-dev
# Make the executable available in the path # Make the executable available in the path
ENV PATH="/app/.venv/bin:$PATH" ENV PATH="/app/.venv/bin:$PATH"
# Default environment variables for the container
ENV MCP_TRANSPORT=http
ENV PORT=8000
ENV HOST=0.0.0.0
# Run the MCP server # Run the MCP server
# By default, strava-mcp uses fastmcp.run() which exposes stdio.
# If you want to run it as an SSE server, you might need to adjust the command.
# For now, we just call the main entrypoint.
ENTRYPOINT ["strava-mcp"] ENTRYPOINT ["strava-mcp"]
+133 -107
View File
@@ -10,10 +10,13 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
- 🛠️ **25+ MCP Tools** covering all major Strava API read endpoints (including athlete profile & stats) - 🛠️ **25+ MCP Tools** covering all major Strava API read endpoints (including athlete profile & stats)
- 💬 **2 MCP Prompts** for structured AI-driven training analysis - 💬 **2 MCP Prompts** for structured AI-driven training analysis
- 🔄 **Fully Automated OAuth** — authentication flow integrated directly as an MCP tool with auto-rotation - 🔄 **Automated OAuth** — authentication flow via a standalone setup script with auto-rotation
- 🐳 **Docker-Ready** highly optimized multi-stage Docker build utilizing `uv` - 🐳 **Multi-Arch Docker** — optimized builds for `linux/amd64` and `linux/arm64` powered by `uv`
- 🏷️ **Dynamic Versioning** — versions are automatically derived from Git tags (powered by `hatch-vcs`)
- 🤖 **Agent-First Design** — includes specific instructions for LLMs on handling European date formats (DD.MM.YYYY)
- 🌐 **Streamable HTTP transport** for broad client compatibility (SSE) - 🌐 **Streamable HTTP transport** for broad client compatibility (SSE)
- 🔒 **Read-only** — no write operations, safe to use with AI agents - 🔒 **Read-only** — no write operations, safe to use with AI agents
- 📝 **Design Decisions** — documented architectural choices in `docs/DESIGN_DECISIONS.md`
--- ---
@@ -24,10 +27,13 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
- [Docker (Recommended)](#docker-recommended) - [Docker (Recommended)](#docker-recommended)
- [Local Python (uv)](#local-python-uv) - [Local Python (uv)](#local-python-uv)
- [Strava API Setup](#strava-api-setup) - [Strava API Setup](#strava-api-setup)
- [Connecting with MCP Inspector](#connecting-with-mcp-inspector) - [Connecting with MCP Clients](#connecting-with-mcp-clients)
- [MCP Primitives](#mcp-primitives) - [MCP Primitives](#mcp-primitives)
- [Project Structure](#project-structure) - [Project Structure](#project-structure)
- [Design Decisions](#design-decisions)
- [CI/CD (Gitea Actions)](#cicd-gitea-actions) - [CI/CD (Gitea Actions)](#cicd-gitea-actions)
- [Development & Testing](#-development--testing)
- [Git Hooks](#git-hooks)
- [Known Strava API Limitations](#known-strava-api-limitations) - [Known Strava API Limitations](#known-strava-api-limitations)
- [Troubleshooting](#troubleshooting) - [Troubleshooting](#troubleshooting)
@@ -37,7 +43,7 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
- A [Strava account](https://www.strava.com) with API access - A [Strava account](https://www.strava.com) with API access
- A [Strava API Application](https://www.strava.com/settings/api) - A [Strava API Application](https://www.strava.com/settings/api)
- **Docker** (for containerized deployment) OR **Python 3.10+** & [uv](https://github.com/astral-sh/uv) (for local execution) - **Docker** (for containerized deployment) OR **Python 3.10+** & [uv](https://github.com/astral-sh/uv)
--- ---
@@ -45,14 +51,14 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
### Docker (Recommended) ### Docker (Recommended)
The project includes a highly optimized, deterministic Dockerfile powered by `uv`. The project includes a multi-arch Docker build (amd64/arm64).
```bash ```bash
# Clone the repository # Clone the repository
git clone https://git.hnrx.net/hnrx/strava-mcp-server.git git clone https://git.hnrx.net/hnrx/strava-mcp-server.git
cd strava-mcp-server cd strava-mcp-server
# Build the image # Build the image locally
docker build -t strava-mcp-server:latest . docker build -t strava-mcp-server:latest .
# Run the container (injecting your .env file) # Run the container (injecting your .env file)
@@ -61,24 +67,29 @@ docker run --rm -p 8000:8000 --env-file .env strava-mcp-server:latest
### Local Python (uv) ### Local Python (uv)
We use `uv` for lightning-fast dependency management and task execution.
```bash ```bash
git clone https://git.hnrx.net/hnrx/strava-mcp-server.git git clone https://git.hnrx.net/hnrx/strava-mcp-server.git
cd strava-mcp-server cd strava-mcp-server
# Install dependencies and start the server # Start the MCP server
uv run strava-mcp uv run server
# Run the OAuth setup script
uv run auth
``` ```
### Run on the fly with `uvx` (No git clone required) ### Run on the fly with `uvx` (No git clone required)
You can run the server directly from the repository without cloning it manually by using `uvx`. `uv` will download it into a temporary isolated environment and execute it: You can run the server directly from the repository without cloning it manually by using `uvx`:
```bash ```bash
# Set up your .env file in the current directory first! # Set up your .env file in the current directory first!
uvx --from git+https://git.hnrx.net/hnrx/strava-mcp-server.git strava-mcp uvx --from git+https://git.hnrx.net/hnrx/strava-mcp-server.git server
``` ```
*(If you are already inside the cloned directory, you can also just run `uvx --from . strava-mcp`)* *(If you are already inside the cloned directory, you can also just run `uvx --from . server`)*
--- ---
@@ -105,27 +116,22 @@ STRAVA_CLIENT_SECRET=your_client_secret_here
### 3. Authenticate (The Magic Way ✨) ### 3. Authenticate (The Magic Way ✨)
You **do not** need to manually fiddle with OAuth tokens. The server includes an interactive MCP tool to handle authentication! The server is designed for zero-touch deployment. You can authorize it **after** it has started.
1. Start the server (`docker run ...` or `uv run strava-mcp`). 1. **Start the server** (it will log a warning that the refresh token is missing, but it will boot!).
2. Connect to the server via an MCP Client (like Claude Desktop or MCP Inspector). 2. **Run the Auth Script**:
3. Call the `get_new_oauth_token` MCP tool. - Run `uv run auth` in your terminal on your local machine.
4. Your browser will open for you to authorize the app. The server will intercept the callback locally, generate your tokens, and automatically save the `STRAVA_REFRESH_TOKEN` to your `.env` file! 3. Your browser will open. Log in and authorize.
4. **Success:** The browser will show you the exact values for your `.env` (or Kubernetes Secret). The script will also automatically update your local `.env` file!
> **Required OAuth Scopes:**
> `activity:read_all,profile:read_all,read`
--- ---
## Connecting with MCP Clients ## Connecting with MCP Clients
The server listens on **port 8000** by default and exposes an SSE endpoint: The server listens on **port 8000** and exposes an SSE endpoint: `http://localhost:8000/mcp`
`http://localhost:8000/mcp`
### Claude Desktop ### Claude Desktop
Add to `claude_desktop_config.json`:
Add to your `claude_desktop_config.json`:
```json ```json
{ {
"mcpServers": { "mcpServers": {
@@ -137,112 +143,132 @@ Add to your `claude_desktop_config.json`:
} }
``` ```
### MCP Inspector
1. Open [MCP Inspector](https://inspector.modelcontextprotocol.io/)
2. Select transport: **Streamable HTTP**
3. Enter URL: `http://localhost:8000/mcp`
4. Click **Connect**
--- ---
## MCP Primitives ## MCP Primitives
### Tools ### Tools
#### 🔐 Authentication | Category | Tools |
| Tool | Description | |----------|-------|
|------|-------------| | 🏃 **Athlete** | `get_athlete_profile`, `get_athlete_stats`, `get_athlete_zones` |
| `get_new_oauth_token` | Starts the interactive browser OAuth2 flow to generate and save your initial Refresh Token | | 🚴 **Activities** | `list_activities`, `get_activity_details`, `get_activity_laps`, `get_activity_zones`, `get_activity_streams` |
| 🏘️ **Clubs** | `get_athlete_clubs`, `get_club_activities`, `get_club_members` |
#### 🏃 Athlete
| Tool | Description |
|------|-------------|
| `get_athlete_profile` | Full athlete profile: name, city, country, follower count, gear list |
| `get_athlete_stats` | Training totals: all-time, year-to-date, and last 4 weeks for runs, rides, and swims |
| `get_athlete_zones` | Heart rate and power zones |
#### 🚴 Activities
| Tool | Description |
|------|-------------|
| `list_activities` | Paginated activity list with optional time range filters |
| `get_activity_details` | Full activity details incl. segment efforts |
| `get_activity_laps` | Lap splits |
| `get_activity_zones` | Heart rate and power zones for a specific activity |
| `get_activity_comments` | Comments on an activity |
| `get_activity_kudoers` | Athletes who gave kudos |
| `get_activity_streams` | Raw GPS/sensor data streams |
*(Note: Additional tools exist for Clubs, Routes, Segments, Segment Efforts, and Gear. See MCP Inspector for full details.)*
### Prompts
Prompts pre-structure AI conversations with the right tool-calling instructions.
- **`analyze_activity`**: Triggers a structured analysis of a specific activity including summary, performance metrics, and key takeaways.
- **`training_summary`**: Generates a training load report for the last N weeks (volume, trends, recommendations).
--- ---
## Project Structure ## Design Decisions
``` For a detailed list of architectural choices, unit standardizations, and LLM-specific optimizations, please refer to:
strava-mcp-server/ 👉 **[docs/DESIGN_DECISIONS.md](docs/DESIGN_DECISIONS.md)**
├── Dockerfile # Multi-stage optimized uv build
├── src/
│ └── strava_mcp_server/ # Installable Python package
│ ├── __init__.py
│ ├── main.py # Server entrypoint → strava-mcp
│ ├── strava_client.py # Strava API client with auto token rotation
│ └── tools/ # Modularized MCP tools directory
│ ├── __init__.py # Tool registry
│ ├── activities.py
│ ├── athlete.py
│ ├── auth.py # OAuth automation flow
│ └── ...
├── .gitea/
│ └── workflows/ # Gitea Actions CI/CD Pipeline
├── tests/
├── pyproject.toml
└── .env
```
--- ---
## CI/CD (Gitea Actions) ## CI/CD (Gitea Actions)
This repository includes a pre-configured Gitea Action (`.gitea/workflows/cicd.yml`) that automatically: Our pipeline (`.gitea/workflows/cicd.yml`) is fully automated:
1. **Lints** the codebase using `ruff` on every push/PR. - **Linting:** Every push/PR is checked with `ruff`.
2. **Builds & Pushes** the Docker container to the local Container Registry (`git.hnrx.net`) upon a successful merge to `main`. - **Multi-Arch Builds:** Builds `amd64` and `arm64` images simultaneously using QEMU and DinD.
- **Smart Tagging:**
--- - Pushes to `main` are tagged as `:latest`.
- Git Tags (e.g., `v1.2.0`) trigger a versioned build and **automatically update the Gitea Release description** with the correct `docker pull` command.
## Known Strava API Limitations
| Endpoint | Status | Reason |
|----------|--------|--------|
| `GET /segments/{id}/leaderboard` | `403 Forbidden` | Requires Strava API partnership |
| `GET /segment_efforts/{id}` | `403 Forbidden` | Requires Strava API partnership |
| `GET /athlete/zones` | `401 Unauthorized` | Requires `profile:read_all` OAuth scope |
> **Workaround for segment efforts:** Use `get_activity_details` to access segment efforts embedded in activity data. The `segment_efforts[]` array contains effort IDs, times, heart rate, power, and PR/KOM ranks.
--- ---
## Troubleshooting ## Troubleshooting
### `[Errno 48] Address already in use` ### `[Errno 48] Address already in use`
Port 8000 is occupied by a previous server process: `lsof -ti :8000 | xargs kill -9`
```bash
lsof -ti :8000 | xargs kill -9
```
### ModuleNotFoundError / iCloud Sync Issues (macOS)
If you are developing locally on macOS and your `strava-mcp-server` directory is located inside `Documents/` or `Desktop/`, **iCloud Drive** will constantly sync and delete files inside your virtual environment (`.venv`), leading to missing packages.
**Solution:** Move the project out of iCloud or rename the folder to end in `.nosync` (e.g. `strava-mcp-server.nosync`).
### 401 Unauthorized ### 401 Unauthorized
Your refresh token has expired or been revoked. Simply run the `get_new_oauth_token` MCP tool again to re-authenticate! Your token expired. Run `uv run auth` to refresh.
---
## 🛠️ Development & Testing
### 1. Local Testing with MCP Inspector
The [MCP Inspector](https://github.com/modelcontextprotocol/inspector) is the best way to test the server without a full LLM client.
**Option A: Test via STDIO (Fastest)**
This runs the server directly in your terminal (perfect for local debugging):
```bash
npx @modelcontextprotocol/inspector uv run server
```
**Option B: Test via SSE (Remote/Docker)**
If the server is already running (e.g., at `http://localhost:8000`):
1. Open [https://inspector.modelcontextprotocol.io/](https://inspector.modelcontextprotocol.io/)
2. Transport: **Streamable HTTP**
3. URL: `http://localhost:8000/mcp`
### 2. Manual SSE Health Check
You can verify if the server is responding to SSE requests using `curl`:
```bash
curl -v -X POST http://localhost:8000/mcp
```
*(It should return an SSE stream starting with `event: endpoint`)*
### 3. Linting & Formatting
We use `ruff` for code quality:
```bash
# Run the check
uv run ruff check src
# Run the formatter
uv run ruff format src
```
### 4. Build Multi-Arch Images
To test if the multi-arch Docker build works locally (requires Docker Buildx):
```bash
docker buildx build --platform linux/amd64,linux/arm64 -t strava-mcp-server:test .
```
---
### 5. Git Hooks
Two hooks are provided in `scripts/hooks/` — install them both after cloning:
```bash
cp scripts/hooks/pre-commit .git/hooks/pre-commit && chmod +x .git/hooks/pre-commit
cp scripts/hooks/pre-push .git/hooks/pre-push && chmod +x .git/hooks/pre-push
```
#### `pre-commit` — Lint & Format check
Runs on staged `.py` files before every commit.
- 🔍 `ruff check` — linting
- 🎨 `ruff format --check` — formatting
- 🔧 Fix: `uv run ruff check --fix` + `uv run ruff format`
- ⚡ Bypass: `git commit --no-verify`
#### `pre-push` — Unit tests
Runs the full unit test suite before every push.
- 🧪 `pytest tests/unit/ -q`
- ⚡ Bypass: `git push --no-verify`
### 6. Unit Tests
Fast, offline unit tests for pure functions and MCP helpers (no Strava API required).
```bash
# Run all unit tests
uv run pytest tests/unit/ -v
# Run with coverage (if pytest-cov is installed)
uv run pytest tests/unit/ --cov=strava_mcp_server
```
**Test coverage:**
| Module | Tests |
|--------|-------|
| `utils.py``parse_iso_to_unix` | `None`, empty, invalid, UTC, offset, date-only |
| `utils.py``format_date_iso` | Normalization, `None`, datetime object, invalid |
| `utils.py``format_date_human` | German format, `None`, datetime object, regex pattern |
| `tools/*``_resource()` | Type, mimeType, URI, JSON validity, audience |
| `tools/*``_user_text()` | Type, text value, audience |
--- ---
+35
View File
@@ -0,0 +1,35 @@
# Design Decisions - Strava MCP Server
This document records key architectural and design decisions made during the development of the Strava MCP Server. It serves as a guide for AI agents and developers to maintain consistency.
## 1. Date and Time Handling
**Decision**: Standardize on ISO 8601 (UTC) for all internal data exchange, tool inputs, and tool outputs.
**Date**: 2026-05-12
**Context**: LLMs often struggle with ambiguous date formats (e.g., DD.MM.YYYY vs. MM/DD/YYYY). International users require a unified format.
**Implementation**:
- All tools accept ISO 8601 strings (`YYYY-MM-DDTHH:MM:SSZ`) for date parameters.
- Tool outputs include a `_date` or similar field with the raw ISO 8601 string.
- A shared utility `src/strava_mcp_server/utils.py` handles parsing and formatting.
- **Human Readability**: While raw data is ISO 8601, markdown summaries presented to the user should use `DD.MM.YYYY HH:MM` for comfort, but the raw data for agent analysis must be standardized.
## 2. Authentication & Token Management
**Decision**: Automate token rotation and prioritize environment-based configuration.
**Context**: Strava tokens expire every 6 hours. Manual refresh is tedious for automated use.
**Implementation**:
- The server checks token expiration before every request.
- Tokens are automatically refreshed and updated in the environment/memory.
- Initial authentication is handled via a separate `auth` script or integrated OAuth flow.
## 3. Data Representation & MCP Annotations
**Decision**: Use dual-content outputs with audience-specific annotations and native MIME typing.
**Context**: To provide a clean user experience while giving the LLM detailed data, we split tool results into two parts and use protocol-level typing.
**Implementation**:
- **User Facing**: Markdown summaries annotated with `audience=["user"]` using `TextContent`.
- **Assistant Facing**: Raw JSON data provided as an `EmbeddedResource` with `mimeType="application/json"` and annotated with `audience=["assistant"]`.
- This ensures the LLM explicitly knows the data format while allowing clients to optimize the user-facing UI.
## 4. Internationalization
**Decision**: Use metric units (meters, km, m/s) internally and provide clear conversion instructions to the LLM.
**Implementation**:
- Strava API returns metric units.
- The LLM is instructed in `main.py` to convert these to human-friendly formats (km, km/h) based on user preference.
+9
View File
@@ -40,11 +40,20 @@ Repository = "https://git.hnrx.net/hnrx/strava-mcp-server"
[project.scripts] [project.scripts]
strava-mcp = "strava_mcp_server.main:main" strava-mcp = "strava_mcp_server.main:main"
strava-mcp-get-token = "strava_mcp_server.get_token:main" strava-mcp-get-token = "strava_mcp_server.get_token:main"
server = "strava_mcp_server.main:main"
auth = "strava_mcp_server.get_token:main"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"ruff>=0.15.12", "ruff>=0.15.12",
"pytest>=8.0",
"pytest-asyncio>=0.24",
"pytest-sugar>=1.1.1",
] ]
[tool.hatch.version] [tool.hatch.version]
source = "vcs" source = "vcs"
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
+35
View File
@@ -0,0 +1,35 @@
#!/bin/sh
# pre-commit hook: runs ruff check (lint) and ruff format --check on staged Python files
# To bypass: git commit --no-verify
STAGED_FILES=$(git diff --cached --name-only --diff-filter=ACM | grep '\.py$')
if [ -z "$STAGED_FILES" ]; then
exit 0
fi
FAILED=0
echo "🔍 Running ruff check (lint)..."
uv run ruff check $STAGED_FILES
if [ $? -ne 0 ]; then
echo " ↳ Fix with: uv run ruff check --fix"
FAILED=1
fi
echo "🎨 Running ruff format --check..."
uv run ruff format --check $STAGED_FILES
if [ $? -ne 0 ]; then
echo " ↳ Fix with: uv run ruff format"
FAILED=1
fi
if [ $FAILED -ne 0 ]; then
echo ""
echo "❌ Pre-commit checks failed. Commit aborted."
echo " To bypass: git commit --no-verify"
exit 1
fi
echo "✅ All checks passed."
exit 0
+18
View File
@@ -0,0 +1,18 @@
#!/bin/sh
# pre-push hook: runs unit tests before every git push
# To bypass: git push --no-verify
echo "🧪 Running unit tests..."
uv run pytest tests/unit/ -q
exit_code=$?
if [ $exit_code -ne 0 ]; then
echo ""
echo "❌ Unit tests failed. Push aborted."
echo " To bypass: git push --no-verify"
exit 1
fi
echo "✅ All unit tests passed."
exit 0
+1
View File
@@ -1,2 +1,3 @@
"""Strava MCP Server""" """Strava MCP Server"""
__version__ = "0.1.0" __version__ = "0.1.0"
+112 -31
View File
@@ -11,6 +11,7 @@ Required scopes:
Usage: uv run get_token.py Usage: uv run get_token.py
""" """
import os import os
import webbrowser import webbrowser
import httpx import httpx
@@ -25,33 +26,98 @@ CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET")
REDIRECT_URI = "http://localhost:8765/callback" REDIRECT_URI = "http://localhost:8765/callback"
SCOPES = "profile:read_all,activity:read_all,activity:read,profile:write" SCOPES = "profile:read_all,activity:read_all,activity:read,profile:write"
# Global to capture the auth code from the callback
auth_code: str | None = None
class CallbackHandler(BaseHTTPRequestHandler): class CallbackHandler(BaseHTTPRequestHandler):
client_id: str = ""
client_secret: str = ""
tokens: dict = {}
error: str | None = None
def do_GET(self): def do_GET(self):
global auth_code
parsed = urlparse(self.path) parsed = urlparse(self.path)
params = parse_qs(parsed.query) params = parse_qs(parsed.query)
if "code" in params: if "code" in params:
auth_code = params["code"][0] code = params["code"][0]
try:
# Exchange code for token synchronously
response = httpx.post(
"https://www.strava.com/oauth/token",
data={
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"grant_type": "authorization_code",
},
)
response.raise_for_status()
self.tokens = response.json()
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "text/html") self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers() self.end_headers()
self.wfile.write(b"""
<html><body style="font-family:sans-serif;text-align:center;padding:40px"> refresh_token = self.tokens.get("refresh_token")
<h2>&#x2705; Authorization successful!</h2>
<p>You can close this window and return to your terminal.</p> self.wfile.write(
</body></html> f"""
""") <html>
<head>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; line-height: 1.6; color: #333; max-width: 800px; margin: 40px auto; padding: 20px; text-align: left; }}
.card {{ background: #f4f7f6; border-radius: 8px; padding: 20px; border-left: 5px solid #fc4c02; margin-top: 20px; }}
pre {{ background: #222; color: #fff; padding: 15px; border-radius: 5px; overflow-x: auto; font-family: "SF Mono", Monaco, Consolas, monospace; font-size: 13px; }}
.success-header {{ text-align: center; margin-bottom: 40px; }}
.success-icon {{ color: #2ecc71; font-size: 48px; display: block; margin-bottom: 10px; }}
h2, h3 {{ color: #fc4c02; }}
.env-label {{ font-weight: bold; color: #666; font-size: 12px; text-transform: uppercase; margin-bottom: 5px; display: block; }}
.copy-hint {{ font-size: 12px; color: #666; font-style: italic; }}
code {{ background: #eee; padding: 2px 4px; border-radius: 4px; }}
</style>
</head>
<body>
<div class="success-header">
<span class="success-icon">&#x2705;</span>
<h2>Authorization successful!</h2>
<p>You have successfully authenticated with Strava. You can now close this window.</p>
</div>
<h3>1. Local Setup (.env)</h3>
<p>Copy the following block into your <code>.env</code> file in the project root:</p>
<div class="card">
<span class="env-label">Your .env content:</span>
<pre>STRAVA_CLIENT_ID={self.client_id}
STRAVA_CLIENT_SECRET={self.client_secret}
STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<h3>2. Kubernetes Setup (Secret)</h3>
<p>If you are deploying this server to Kubernetes, run the following command to create the required Secret:</p>
<div class="card">
<span class="env-label">Kubectl Command:</span>
<pre>kubectl create secret generic strava-mcp-server-secret \\
--from-literal=STRAVA_CLIENT_ID={self.client_id} \\
--from-literal=STRAVA_CLIENT_SECRET={self.client_secret} \\
--from-literal=STRAVA_REFRESH_TOKEN={refresh_token}</pre>
</div>
<p style="margin-top: 40px; font-size: 14px; color: #666; text-align: center;">
&mdash; Strava MCP Server Authorization Helper &mdash;
</p>
</body>
</html>
""".encode("utf-8")
)
except Exception as e:
self.error = str(e)
self.send_response(500)
self.end_headers()
self.wfile.write(f"Error exchanging token: {e}".encode())
else: else:
error = params.get("error", ["unknown"])[0] error_msg = params.get("error", ["unknown"])[0]
self.send_response(400) self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers() self.end_headers()
self.wfile.write(f"<html><body><h2>&#x274C; Error: {error}</h2></body></html>".encode()) self.wfile.write(f"Error: {error_msg}".encode())
def log_message(self, format, *args): def log_message(self, format, *args):
pass # Suppress server logs pass # Suppress server logs
@@ -71,6 +137,12 @@ def main():
f"&scope={SCOPES}" f"&scope={SCOPES}"
) )
# Configure handler
CallbackHandler.client_id = CLIENT_ID
CallbackHandler.client_secret = CLIENT_SECRET
CallbackHandler.tokens = {}
CallbackHandler.error = None
print("=" * 60) print("=" * 60)
print(" Strava OAuth2 Authorization") print(" Strava OAuth2 Authorization")
print("=" * 60) print("=" * 60)
@@ -85,26 +157,15 @@ def main():
server = HTTPServer(("localhost", 8765), CallbackHandler) server = HTTPServer(("localhost", 8765), CallbackHandler)
server.handle_request() # Handle exactly one request (the callback) server.handle_request() # Handle exactly one request (the callback)
if not auth_code: if CallbackHandler.error:
print("No authorization code received.") print(f"Token exchange failed: {CallbackHandler.error}")
return return
print("\nExchanging authorization code for tokens...") if not CallbackHandler.tokens:
response = httpx.post( print("❌ No tokens received.")
"https://www.strava.com/oauth/token",
data={
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"code": auth_code,
"grant_type": "authorization_code",
},
)
if response.status_code != 200:
print(f"❌ Token exchange failed: {response.status_code} {response.text}")
return return
data = response.json() data = CallbackHandler.tokens
refresh_token = data["refresh_token"] refresh_token = data["refresh_token"]
athlete = data.get("athlete", {}) athlete = data.get("athlete", {})
@@ -120,6 +181,26 @@ def main():
print(f"STRAVA_REFRESH_TOKEN={refresh_token}") print(f"STRAVA_REFRESH_TOKEN={refresh_token}")
print("-" * 40) print("-" * 40)
# Optional: Automatically update .env if it exists
try:
env_path = ".env"
if os.path.exists(env_path):
with open(env_path, "r") as f:
lines = f.readlines()
with open(env_path, "w") as f:
found = False
for line in lines:
if line.startswith("STRAVA_REFRESH_TOKEN="):
f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n")
found = True
else:
f.write(line)
if not found:
f.write(f"\nSTRAVA_REFRESH_TOKEN={refresh_token}\n")
print("Successfully updated your .env file!")
except Exception as e:
print(f"Could not automatically update .env: {e}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()
+19 -3
View File
@@ -26,8 +26,10 @@ def validate_credentials() -> None:
sys.exit(1) sys.exit(1)
if not os.getenv("STRAVA_REFRESH_TOKEN"): if not os.getenv("STRAVA_REFRESH_TOKEN"):
print("️ No STRAVA_REFRESH_TOKEN found. Server starting in unauthenticated mode.") print(
print(" Use the 'get_new_oauth_token' tool via MCP to authenticate.") "️ No STRAVA_REFRESH_TOKEN found. Server starting in unauthenticated mode."
)
print(" Run 'uv run auth' on your local machine to authenticate.")
def main() -> None: def main() -> None:
@@ -42,7 +44,13 @@ def main() -> None:
mcp = FastMCP( mcp = FastMCP(
"Strava MCP Server", "Strava MCP Server",
instructions="Dates returned by this server are generally in ISO-8601 (UTC) or formatted as DD.MM.YYYY HH:MM. Always present dates, times, and durations to the user in a natural, human-readable format appropriate for their language.", instructions="""
IMPORTANT ON DATE/TIME:
- Always use ISO 8601 (UTC) for date/time inputs (YYYY-MM-DDTHH:MM:SSZ).
- This server returns dates in ISO 8601 (UTC).
- When presenting to the user, you may format dates naturally in their local language, but use the raw ISO data for all internal logic and tool calls.
- Distance is in meters (convert to km for users), elevation in meters, and speed in m/s (convert to km/h or pace).
""".strip(),
host=host, host=host,
port=port, port=port,
streamable_http_path="/mcp", streamable_http_path="/mcp",
@@ -56,12 +64,20 @@ def main() -> None:
register_tools(mcp, strava) register_tools(mcp, strava)
# Check transport mode from environment (Default to stdio for local dev)
transport = os.getenv("MCP_TRANSPORT", "stdio")
if transport == "http":
# Run in Streamable HTTP mode (standard for Docker, K8s and OpenWebUI)
print(f"🚀 Starting Strava MCP Server on http://{host}:{port}") print(f"🚀 Starting Strava MCP Server on http://{host}:{port}")
print(f" MCP endpoint: http://{host}:{port}/mcp (Streamable HTTP)") print(f" MCP endpoint: http://{host}:{port}/mcp (Streamable HTTP)")
try: try:
mcp.run(transport="streamable-http") mcp.run(transport="streamable-http")
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
else:
# Run in STDIO mode (default for local testing and Claude Desktop)
mcp.run(transport="stdio")
if __name__ == "__main__": if __name__ == "__main__":
+60 -20
View File
@@ -37,7 +37,9 @@ class StravaClient:
async def get_valid_token(self) -> str: async def get_valid_token(self) -> str:
"""Returns a valid access token, refreshing it if necessary.""" """Returns a valid access token, refreshing it if necessary."""
if not self.refresh_token: if not self.refresh_token:
raise ValueError("No Strava refresh token found. Please run the 'get_new_oauth_token' MCP tool to authenticate first.") raise ValueError(
"No Strava refresh token found. Please run 'uv run auth' on your local machine to authenticate first."
)
if not self.access_token or time.time() > self.expires_at - 60: if not self.access_token or time.time() > self.expires_at - 60:
await self._refresh_access_token() await self._refresh_access_token()
@@ -99,15 +101,27 @@ class StravaClient:
"""Gets the heart rate and power zones for a specific activity.""" """Gets the heart rate and power zones for a specific activity."""
return await self._get(f"activities/{activity_id}/zones") return await self._get(f"activities/{activity_id}/zones")
async def get_activity_comments(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_activity_comments(
self, activity_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets comments on a specific activity.""" """Gets comments on a specific activity."""
return await self._get(f"activities/{activity_id}/comments", params={"page": page, "per_page": per_page}) return await self._get(
f"activities/{activity_id}/comments",
params={"page": page, "per_page": per_page},
)
async def get_activity_kudoers(self, activity_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_activity_kudoers(
self, activity_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets athletes who kudoed a specific activity.""" """Gets athletes who kudoed a specific activity."""
return await self._get(f"activities/{activity_id}/kudos", params={"page": page, "per_page": per_page}) return await self._get(
f"activities/{activity_id}/kudos",
params={"page": page, "per_page": per_page},
)
async def get_activity_streams(self, activity_id: int, keys: List[str]) -> Dict[str, Any]: async def get_activity_streams(
self, activity_id: int, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific activity.""" """Gets data streams for a specific activity."""
return await self._get( return await self._get(
f"activities/{activity_id}/streams", f"activities/{activity_id}/streams",
@@ -116,28 +130,45 @@ class StravaClient:
# ── Clubs ───────────────────────────────────────────────────────────────── # ── Clubs ─────────────────────────────────────────────────────────────────
async def get_athlete_clubs(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_athlete_clubs(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets clubs the current athlete belongs to.""" """Gets clubs the current athlete belongs to."""
return await self._get("athlete/clubs", params={"page": page, "per_page": per_page}) return await self._get(
"athlete/clubs", params={"page": page, "per_page": per_page}
)
async def get_club(self, club_id: int) -> Dict[str, Any]: async def get_club(self, club_id: int) -> Dict[str, Any]:
"""Gets a specific club by ID.""" """Gets a specific club by ID."""
return await self._get(f"clubs/{club_id}") return await self._get(f"clubs/{club_id}")
async def get_club_activities(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_club_activities(
self, club_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets recent activities from a club.""" """Gets recent activities from a club."""
return await self._get(f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page}) return await self._get(
f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page}
)
async def get_club_members(self, club_id: int, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_club_members(
self, club_id: int, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets members of a club.""" """Gets members of a club."""
return await self._get(f"clubs/{club_id}/members", params={"page": page, "per_page": per_page}) return await self._get(
f"clubs/{club_id}/members", params={"page": page, "per_page": per_page}
)
# ── Routes ──────────────────────────────────────────────────────────────── # ── Routes ────────────────────────────────────────────────────────────────
async def get_routes_by_athlete(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_routes_by_athlete(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets routes created by the current athlete.""" """Gets routes created by the current athlete."""
athlete = await self.get_athlete() athlete = await self.get_athlete()
return await self._get(f"athletes/{athlete['id']}/routes", params={"page": page, "per_page": per_page}) return await self._get(
f"athletes/{athlete['id']}/routes",
params={"page": page, "per_page": per_page},
)
async def get_route_by_id(self, route_id: str) -> Dict[str, Any]: async def get_route_by_id(self, route_id: str) -> Dict[str, Any]:
"""Gets a specific route by its ID.""" """Gets a specific route by its ID."""
@@ -153,19 +184,26 @@ class StravaClient:
"""Gets a specific segment by ID.""" """Gets a specific segment by ID."""
return await self._get(f"segments/{segment_id}") return await self._get(f"segments/{segment_id}")
async def get_starred_segments(self, page: int = 1, per_page: int = 30) -> List[Dict[str, Any]]: async def get_starred_segments(
self, page: int = 1, per_page: int = 30
) -> List[Dict[str, Any]]:
"""Gets segments starred by the current athlete.""" """Gets segments starred by the current athlete."""
return await self._get("segments/starred", params={"page": page, "per_page": per_page}) return await self._get(
"segments/starred", params={"page": page, "per_page": per_page}
)
async def explore_segments(self, bounds: str, activity_type: str = "riding") -> Dict[str, Any]: async def explore_segments(
self, bounds: str, activity_type: str = "riding"
) -> Dict[str, Any]:
"""Explores segments within a given bounding box.""" """Explores segments within a given bounding box."""
return await self._get( return await self._get(
"segments/explore", "segments/explore",
params={"bounds": bounds, "activity_type": activity_type}, params={"bounds": bounds, "activity_type": activity_type},
) )
async def get_segment_streams(
async def get_segment_streams(self, segment_id: int, keys: List[str]) -> Dict[str, Any]: self, segment_id: int, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific segment.""" """Gets data streams for a specific segment."""
return await self._get( return await self._get(
f"segments/{segment_id}/streams", f"segments/{segment_id}/streams",
@@ -193,7 +231,9 @@ class StravaClient:
params["end_date_local"] = end_date_local params["end_date_local"] = end_date_local
return await self._get("segment_efforts", params=params) return await self._get("segment_efforts", params=params)
async def get_segment_effort_streams(self, effort_id: str, keys: List[str]) -> Dict[str, Any]: async def get_segment_effort_streams(
self, effort_id: str, keys: List[str]
) -> Dict[str, Any]:
"""Gets data streams for a specific segment effort.""" """Gets data streams for a specific segment effort."""
return await self._get( return await self._get(
f"segment_efforts/{effort_id}/streams", f"segment_efforts/{effort_id}/streams",
+2 -2
View File
@@ -3,6 +3,7 @@ MCP Tool definitions for the Strava MCP Server.
Register all tools by calling register_tools(mcp, strava). Register all tools by calling register_tools(mcp, strava).
""" """
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
@@ -14,7 +15,7 @@ from . import segments
from . import segment_efforts from . import segment_efforts
from . import gear from . import gear
from . import prompts from . import prompts
from . import auth
def register_tools(mcp: FastMCP, strava: StravaClient) -> None: def register_tools(mcp: FastMCP, strava: StravaClient) -> None:
"""Register all available tools and prompts.""" """Register all available tools and prompts."""
@@ -26,4 +27,3 @@ def register_tools(mcp: FastMCP, strava: StravaClient) -> None:
segment_efforts.register(mcp, strava) segment_efforts.register(mcp, strava)
gear.register(mcp, strava) gear.register(mcp, strava)
prompts.register(mcp, strava) prompts.register(mcp, strava)
auth.register(mcp, strava)
+182 -46
View File
@@ -1,7 +1,33 @@
import json import json
from mcp.server.fastmcp import FastMCP, Context from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import (
parse_iso_to_unix,
format_date_iso,
format_date_human,
)
def _resource(uri: str, data) -> EmbeddedResource:
"""Helper: return an assistant-facing EmbeddedResource with application/json."""
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri,
mimeType="application/json",
text=json.dumps(data, indent=2),
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
"""Helper: return a user-facing TextContent."""
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
@@ -9,16 +35,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
ctx: Context, ctx: Context,
limit: int = 10, limit: int = 10,
page: int = 1, page: int = 1,
before: int | None = None, before: str | None = None,
after: int | None = None, after: str | None = None,
) -> list[TextContent]: ):
""" """
List recent Strava activities for the authenticated user. List recent Strava activities for the authenticated user.
:param limit: Number of activities to return per page (default 10, max 200). :param limit: Number of activities to return per page (default 10, max 200).
:param page: Page number for pagination (default 1). :param page: Page number for pagination (default 1).
:param before: Unix timestamp — only return activities before this time. :param before: ISO 8601 date string (e.g., '2024-05-01T00:00:00Z') — only return activities before this time.
:param after: Unix timestamp — only return activities after this time. :param after: ISO 8601 date string (e.g., '2024-04-01T00:00:00Z') — only return activities after this time.
Example: 1704067200 = 2024-01-01 00:00:00 UTC.
""" """
try: try:
await ctx.info(f"Fetching activities (Page {page}, Limit {limit})...") await ctx.info(f"Fetching activities (Page {page}, Limit {limit})...")
@@ -26,50 +51,51 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
activities = await strava.list_activities( activities = await strava.list_activities(
limit=min(limit, 200), limit=min(limit, 200),
page=page, page=page,
before=before, before=parse_iso_to_unix(before),
after=after, after=parse_iso_to_unix(after),
) )
from datetime import datetime
def format_date(d_str):
if not d_str:
return "N/A"
try:
d = datetime.fromisoformat(d_str.replace('Z', '+00:00'))
return f"{d.day:02d}.{d.month:02d}.{d.year} {d.hour:02d}:{d.minute:02d}"
except Exception:
return d_str
essential_data = [] essential_data = []
for a in activities: for a in activities:
essential_data.append({ start_date_raw = a.get("start_date")
essential_data.append(
{
"id": a["id"], "id": a["id"],
"name": a["name"], "name": a["name"],
"sport_type": a.get("sport_type") or a.get("type"), "sport_type": a.get("sport_type") or a.get("type"),
"start_date": format_date(a.get("start_date")), "start_date": format_date_iso(start_date_raw),
"start_date_local": format_date_human(start_date_raw),
"distance": f"{a.get('distance', 0) / 1000:.2f} km", "distance": f"{a.get('distance', 0) / 1000:.2f} km",
"moving_time": f"{a.get('moving_time', 0) / 60:.1f} min", "moving_time": f"{a.get('moving_time', 0) / 60:.1f} min",
"total_elevation_gain": f"{a.get('total_elevation_gain', 0):.0f} m", "total_elevation_gain": f"{a.get('total_elevation_gain', 0):.0f} m",
"average_heartrate": a.get("average_heartrate"), "average_heartrate": a.get("average_heartrate"),
"gear_id": a.get("gear_id"), "gear_id": a.get("gear_id"),
}) }
)
if not essential_data: if not essential_data:
markdown_summary = "### 📭 Keine Aktivitäten in diesem Zeitraum gefunden." markdown_summary = (
"### 📭 Keine Aktivitäten in diesem Zeitraum gefunden."
)
else: else:
markdown_summary = f"### 🚴 Aktivitäten (Seite {page})\n" markdown_summary = f"### 🚴 Aktivitäten (Seite {page})\n"
markdown_summary += "| Datum | Sport | Name | Distanz | Zeit | Höhenmeter | Ø HR |\n" markdown_summary += (
markdown_summary += "|-------|-------|------|---------|------|------------|------|\n" "| Datum | Sport | Name | Distanz | Zeit | Höhenmeter | Ø HR |\n"
)
markdown_summary += (
"|-------|-------|------|---------|------|------------|------|\n"
)
for a in essential_data: for a in essential_data:
hr = f"{a['average_heartrate']:.0f} bpm" if a['average_heartrate'] else "-" hr = (
markdown_summary += f"| {a['start_date']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n" f"{a['average_heartrate']:.0f} bpm"
if a["average_heartrate"]
else "-"
)
markdown_summary += f"| {a['start_date_local']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
return [ return [
TextContent(type="text", text=markdown_summary.strip()), _user_text(markdown_summary.strip()),
TextContent( _resource("internal://activities/list", essential_data),
type="text",
text=f"Raw Data for Analysis: {json.dumps(essential_data, indent=2)}"
)
] ]
except Exception as e: except Exception as e:
error_msg = f"Error listing activities: {str(e)}" error_msg = f"Error listing activities: {str(e)}"
@@ -94,9 +120,61 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
segment["id"] = str(segment["id"]) segment["id"] = str(segment["id"])
if "id" in activity: if "id" in activity:
activity["id"] = str(activity["id"]) activity["id"] = str(activity["id"])
return activity
name = activity.get("name", "N/A")
sport = activity.get("sport_type") or activity.get("type", "N/A")
date = format_date_human(activity.get("start_date"))
dist = f"{activity.get('distance', 0) / 1000:.2f} km"
time = f"{activity.get('moving_time', 0) / 60:.1f} min"
elev = f"{activity.get('total_elevation_gain', 0):.0f} m"
avg_hr = (
f"{activity.get('average_heartrate', 0):.0f} bpm"
if activity.get("average_heartrate")
else "N/A"
)
max_hr = (
f"{activity.get('max_heartrate', 0):.0f} bpm"
if activity.get("max_heartrate")
else "N/A"
)
avg_spd = (
f"{activity.get('average_speed', 0) * 3.6:.1f} km/h"
if activity.get("average_speed")
else "N/A"
)
avg_w = (
f"{activity.get('average_watts', 0):.0f} W"
if activity.get("average_watts")
else "N/A"
)
gear = activity.get("gear_id") or "N/A"
n_efforts = len(activity.get("segment_efforts", []))
markdown_summary = f"""### 🏃 Aktivität: {name}
| Feld | Wert |
|------|------|
| Sport | {sport} |
| Datum | {date} |
| Distanz | {dist} |
| Zeit | {time} |
| Höhenmeter | {elev} |
| Ø Herzfrequenz | {avg_hr} |
| Max Herzfrequenz | {max_hr} |
| Ø Geschwindigkeit | {avg_spd} |
| Ø Leistung | {avg_w} |
| Ausrüstung | {gear} |
| Segment-Efforts | {n_efforts} |"""
return [
_user_text(markdown_summary.strip()),
_resource(f"internal://activities/{activity_id}", activity),
]
except Exception as e: except Exception as e:
return f"Error fetching activity details: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching activity details: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def get_activity_comments(activity_id: int, limit: int = 30): async def get_activity_comments(activity_id: int, limit: int = 30):
@@ -107,17 +185,27 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
comments = await strava.get_activity_comments(activity_id, per_page=limit) comments = await strava.get_activity_comments(activity_id, per_page=limit)
return [ data = [
{ {
"id": c.get("id"), "id": c.get("id"),
"text": c.get("text"), "text": c.get("text"),
"athlete": f"{c.get('athlete', {}).get('firstname')} {c.get('athlete', {}).get('lastname')}", "athlete": f"{c.get('athlete', {}).get('firstname')} {c.get('athlete', {}).get('lastname')}",
"created_at": c.get("created_at"), "created_at": format_date_iso(c.get("created_at")),
} }
for c in comments for c in comments
] ]
if not data:
md = "### 💬 Keine Kommentare vorhanden."
else:
md = f"### 💬 Kommentare ({len(data)})\n"
for c in data:
md += f"- **{c['athlete']}** ({format_date_human(c['created_at'])}): {c['text']}\n"
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/comments", data),
]
except Exception as e: except Exception as e:
return f"Error fetching comments: {str(e)}" return [TextContent(type="text", text=f"Error fetching comments: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_activity_kudoers(activity_id: int, limit: int = 30): async def get_activity_kudoers(activity_id: int, limit: int = 30):
@@ -128,7 +216,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
kudoers = await strava.get_activity_kudoers(activity_id, per_page=limit) kudoers = await strava.get_activity_kudoers(activity_id, per_page=limit)
return [ data = [
{ {
"id": k.get("id"), "id": k.get("id"),
"name": f"{k.get('firstname')} {k.get('lastname')}", "name": f"{k.get('firstname')} {k.get('lastname')}",
@@ -137,8 +225,20 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for k in kudoers for k in kudoers
] ]
if not data:
md = "### 👍 Noch keine Kudos."
else:
md = f"### 👍 Kudos ({len(data)})\n"
md += "| Name | Ort |\n|------|-----|\n"
for k in data:
loc = ", ".join(filter(None, [k["city"], k["country"]])) or "N/A"
md += f"| {k['name']} | {loc} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/kudoers", data),
]
except Exception as e: except Exception as e:
return f"Error fetching kudoers: {str(e)}" return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_laps_by_activity_id(activity_id: int): async def get_laps_by_activity_id(activity_id: int):
@@ -149,7 +249,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
laps = await strava.get_activity_laps(activity_id) laps = await strava.get_activity_laps(activity_id)
return [ data = [
{ {
"lap_index": lap.get("lap_index"), "lap_index": lap.get("lap_index"),
"name": lap.get("name"), "name": lap.get("name"),
@@ -164,8 +264,25 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for lap in laps for lap in laps
] ]
if not data:
md = "### 🔄 Keine Runden gefunden."
else:
md = f"### 🔄 Runden ({len(data)})\n"
md += "| # | Distanz | Zeit | Ø Speed | Ø HR |\n"
md += "|---|---------|------|---------|------|\n"
for lap in data:
hr = (
f"{lap['average_heartrate']:.0f} bpm"
if lap["average_heartrate"]
else "-"
)
md += f"| {lap['lap_index']} | {lap['distance']} | {lap['moving_time']} | {lap['average_speed']} | {hr} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/laps", data),
]
except Exception as e: except Exception as e:
return f"Error fetching laps: {str(e)}" return [TextContent(type="text", text=f"Error fetching laps: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_zones_by_activity_id(activity_id: int): async def get_zones_by_activity_id(activity_id: int):
@@ -176,7 +293,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
zones = await strava.get_activity_zones(activity_id) zones = await strava.get_activity_zones(activity_id)
result = [] data = []
md = "### 💓 Zonen-Verteilung\n\n"
for zone in zones: for zone in zones:
buckets = [ buckets = [
{ {
@@ -187,17 +305,35 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for i, b in enumerate(zone.get("distribution_buckets", [])) for i, b in enumerate(zone.get("distribution_buckets", []))
] ]
result.append({ zone_data = {
"type": zone.get("type", "unknown"), "type": zone.get("type", "unknown"),
"sensor_based": zone.get("sensor_based", False), "sensor_based": zone.get("sensor_based", False),
"score": zone.get("score"), "score": zone.get("score"),
"custom_zones": zone.get("custom_zones", False), "custom_zones": zone.get("custom_zones", False),
"points": zone.get("points"), "points": zone.get("points"),
"distribution_buckets": buckets, "distribution_buckets": buckets,
}) }
return result data.append(zone_data)
label = (
"Herzfrequenz"
if zone_data["type"] == "heartrate"
else "Leistung (Power)"
)
md += f"#### {label}\n| Zone | Bereich | Zeit |\n|------|---------|------|\n"
for b in buckets:
max_val = "max" if b["max"] == -1 else str(b["max"])
md += f"| {b['zone']} | {b['min']} {max_val} | {b['time_in_zone']} |\n"
md += "\n"
return [
_user_text(md.strip()),
_resource(f"internal://activities/{activity_id}/zones", data),
]
except Exception as e: except Exception as e:
return f"Error fetching activity zones: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching activity zones: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def get_activity_streams( async def get_activity_streams(
+136 -53
View File
@@ -1,33 +1,31 @@
import json import json
from mcp.server.fastmcp import FastMCP, Context from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import format_date_iso, format_date_human
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
async def get_athlete_profile(ctx: Context) -> list[TextContent]: async def get_athlete_profile(ctx: Context):
""" """
Get the authenticated Strava athlete's profile. Get the authenticated Strava athlete's profile.
Returns name, city, country, follower count, and other profile details. Returns name, city, country, follower count, and other profile details.
""" """
try: try:
await ctx.info("Fetching athlete profile...") await ctx.info("Fetching athlete profile...")
athlete = await strava.get_athlete() athlete = await strava.get_athlete()
from datetime import datetime location_parts = [
p
def format_date(d_str): for p in (
if not d_str: athlete.get("city"),
return "N/A" athlete.get("state"),
try: athlete.get("country"),
d = datetime.fromisoformat(d_str.replace('Z', '+00:00')) )
return f"{d.day}.{d.month}.{d.year}" if p
except Exception: ]
return d_str
location_parts = [p for p in (athlete.get('city'), athlete.get('state'), athlete.get('country')) if p]
location = ", ".join(location_parts) if location_parts else "N/A" location = ", ".join(location_parts) if location_parts else "N/A"
essential_data = { essential_data = {
@@ -40,32 +38,41 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"measurement_units": athlete.get("measurement_preference"), "measurement_units": athlete.get("measurement_preference"),
"is_premium": athlete.get("premium", False), "is_premium": athlete.get("premium", False),
"profile_medium": athlete.get("profile_medium"), "profile_medium": athlete.get("profile_medium"),
"created_at": athlete.get("created_at"), "created_at": format_date_iso(athlete.get("created_at")),
"updated_at": athlete.get("updated_at"), "updated_at": format_date_iso(athlete.get("updated_at")),
"bio": athlete.get("bio"), "bio": athlete.get("bio"),
"follower_count": athlete.get("follower_count"), "follower_count": athlete.get("follower_count"),
"friend_count": athlete.get("friend_count"), "friend_count": athlete.get("friend_count"),
} }
markdown_summary = f""" markdown_summary = f"""
👤 **Profile for {essential_data['name']}** (ID: {essential_data['id']}) 👤 **Profile for {essential_data["name"]}** (ID: {essential_data["id"]})
- Username: {essential_data['username'] or 'N/A'} - Username: {essential_data["username"] or "N/A"}
- Location: {essential_data['location']} - Location: {essential_data["location"]}
- Sex: {essential_data['sex'] or 'N/A'} - Sex: {essential_data["sex"] or "N/A"}
- Weight: {essential_data['weight'] or 'N/A'} kg - Weight: {essential_data["weight"] or "N/A"} kg
- Measurement Units: {essential_data['measurement_units'] or 'N/A'} - Measurement Units: {essential_data["measurement_units"] or "N/A"}
- Strava Summit Member: {'Yes' if essential_data['is_premium'] else 'No'} - Strava Summit Member: {"Yes" if essential_data["is_premium"] else "No"}
- Profile Image (Medium): {essential_data['profile_medium'] or 'N/A'} - Profile Image (Medium): {essential_data["profile_medium"] or "N/A"}
- Joined Strava: {format_date(essential_data['created_at'])} - Joined Strava: {format_date_human(essential_data["created_at"])}
- Last Updated: {format_date(essential_data['updated_at'])} - Last Updated: {format_date_human(essential_data["updated_at"])}
""".strip() """.strip()
return [ return [
TextContent(type="text", text=markdown_summary),
TextContent( TextContent(
type="text", type="text",
text=f"Raw Data for Analysis: {json.dumps(essential_data, indent=2)}" text=markdown_summary,
) annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/profile",
mimeType="application/json",
text=json.dumps(essential_data, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
] ]
except Exception as e: except Exception as e:
@@ -74,24 +81,68 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
return [TextContent(type="text", text=error_msg)] return [TextContent(type="text", text=error_msg)]
@mcp.tool() @mcp.tool()
async def get_athlete_zones(): async def get_athlete_zones(ctx: Context):
""" """
Get the heart rate and power zones configured for the authenticated athlete. Get the heart rate and power zones configured for the authenticated athlete.
Returns zone boundaries for both heart rate and power (if a power meter is configured). Returns zone boundaries for both heart rate and power (if a power meter is configured).
""" """
try: try:
return await strava.get_athlete_zones() await ctx.info("Fetching athlete zones...")
zones = await strava.get_athlete_zones()
markdown_summary = "### 💓 Trainingszonen\n\n"
# Heart Rate Zones
hr_zones = zones.get("heart_rate", {}).get("zones", [])
if hr_zones:
markdown_summary += "#### Herzfrequenz-Zonen\n"
markdown_summary += "| Zone | Bereich (bpm) |\n"
markdown_summary += "|------|---------------|\n"
for i, z in enumerate(hr_zones):
markdown_summary += f"| {i + 1} | {z.get('min')} - {z.get('max') if z.get('max') != -1 else 'max'} |\n"
markdown_summary += "\n"
# Power Zones
power_zones = zones.get("power", {}).get("zones", [])
if power_zones:
markdown_summary += "#### Leistungs-Zonen (Power)\n"
markdown_summary += "| Zone | Bereich (W) |\n"
markdown_summary += "|------|-------------|\n"
for i, z in enumerate(power_zones):
markdown_summary += f"| {i + 1} | {z.get('min')} - {z.get('max') if z.get('max') != -1 else 'max'} |\n"
if not hr_zones and not power_zones:
markdown_summary = "⚠️ Keine Trainingszonen konfiguriert."
return [
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/zones",
mimeType="application/json",
text=json.dumps(zones, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
]
except Exception as e: except Exception as e:
return f"Error fetching athlete zones: {str(e)}" error_msg = f"Error fetching athlete zones: {str(e)}"
await ctx.error(error_msg)
return [TextContent(type="text", text=error_msg)]
@mcp.tool() @mcp.tool()
async def get_athlete_stats(): async def get_athlete_stats(ctx: Context):
""" """
Get cumulative training statistics for the authenticated Strava athlete. Get cumulative training statistics for the authenticated Strava athlete.
Includes all-time, year-to-date, and recent (4-week) totals for runs, rides, and swims. Includes all-time, year-to-date, and recent (4-week) totals for runs, rides, and swims.
""" """
import json
try: try:
await ctx.info("Fetching athlete statistics...")
stats = await strava.get_athlete_stats() stats = await strava.get_athlete_stats()
def fmt_sport(s: dict) -> dict: def fmt_sport(s: dict) -> dict:
@@ -102,23 +153,55 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"elevation_gain": f"{s.get('elevation_gain', 0):.0f} m", "elevation_gain": f"{s.get('elevation_gain', 0):.0f} m",
} }
result = { # Prepare structured data for Markdown
"all_time": { all_time = {
"runs": fmt_sport(stats.get("all_run_totals", {})), "Laufen": fmt_sport(stats.get("all_run_totals", {})),
"rides": fmt_sport(stats.get("all_ride_totals", {})), "Radfahren": fmt_sport(stats.get("all_ride_totals", {})),
"swims": fmt_sport(stats.get("all_swim_totals", {})), "Schwimmen": fmt_sport(stats.get("all_swim_totals", {})),
},
"ytd": {
"runs": fmt_sport(stats.get("ytd_run_totals", {})),
"rides": fmt_sport(stats.get("ytd_ride_totals", {})),
"swims": fmt_sport(stats.get("ytd_swim_totals", {})),
},
"recent_4_weeks": {
"runs": fmt_sport(stats.get("recent_run_totals", {})),
"rides": fmt_sport(stats.get("recent_ride_totals", {})),
"swims": fmt_sport(stats.get("recent_swim_totals", {})),
},
} }
return json.dumps(result, indent=2) ytd = {
"Laufen": fmt_sport(stats.get("ytd_run_totals", {})),
"Radfahren": fmt_sport(stats.get("ytd_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("ytd_swim_totals", {})),
}
recent = {
"Laufen": fmt_sport(stats.get("recent_run_totals", {})),
"Radfahren": fmt_sport(stats.get("recent_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("recent_swim_totals", {})),
}
markdown_summary = "### 📈 Trainingsstatistiken\n\n"
def create_table(title: str, data: dict):
tbl = f"#### {title}\n"
tbl += "| Sport | Aktivitäten | Distanz | Zeit | Höhenmeter |\n"
tbl += "|-------|-------------|---------|------|------------|\n"
for sport, s in data.items():
if s["count"] > 0:
tbl += f"| {sport} | {s['count']} | {s['distance']} | {s['moving_time']} | {s['elevation_gain']} |\n"
return tbl + "\n"
markdown_summary += create_table("Letzte 4 Wochen", recent)
markdown_summary += create_table("Dieses Jahr (YTD)", ytd)
markdown_summary += create_table("Gesamt", all_time)
return [
TextContent(
type="text",
text=markdown_summary.strip(),
annotations=Annotations(audience=["user"]),
),
EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri="internal://athlete/stats",
mimeType="application/json",
text=json.dumps(stats, indent=2),
),
annotations=Annotations(audience=["assistant"]),
),
]
except Exception as e: except Exception as e:
return f"Error fetching athlete stats: {str(e)}" error_msg = f"Error fetching athlete stats: {str(e)}"
await ctx.error(error_msg)
return [TextContent(type="text", text=error_msg)]
-121
View File
@@ -1,121 +0,0 @@
import os
import webbrowser
import httpx
import asyncio
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent
auth_code: str | None = None
class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
global auth_code
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
if "code" in params:
auth_code = params["code"][0]
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(b"""
<html><body style="font-family:sans-serif;text-align:center;padding:40px">
<h2>&#x2705; Authorization successful!</h2>
<p>You can close this window and return to your terminal/chat.</p>
</body></html>
""")
else:
error = params.get("error", ["unknown"])[0]
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(f"<html><body><h2>&#x274C; Error: {error}</h2></body></html>".encode())
def log_message(self, format, *args):
pass # Suppress server logs
def register(mcp: FastMCP, strava) -> None:
@mcp.tool()
async def get_new_oauth_token(ctx: Context) -> list[TextContent]:
"""
Start the interactive Strava OAuth2 authorization flow.
This opens a browser window for the user to log in and authorize the app.
It then intercepts the redirect locally, obtains the token, and returns the tokens.
"""
global auth_code
auth_code = None
client_id = os.getenv("STRAVA_CLIENT_ID")
client_secret = os.getenv("STRAVA_CLIENT_SECRET")
if not client_id or not client_secret:
return [TextContent(type="text", text="Error: Missing STRAVA_CLIENT_ID or STRAVA_CLIENT_SECRET in .env")]
redirect_uri = "http://localhost:8765/callback"
scopes = "profile:read_all,activity:read_all,activity:read,profile:write"
auth_url = (
f"https://www.strava.com/oauth/authorize"
f"?client_id={client_id}"
f"&redirect_uri={redirect_uri}"
f"&response_type=code"
f"&approval_prompt=force"
f"&scope={scopes}"
)
await ctx.info("Opening browser for Strava Authorization...")
webbrowser.open(auth_url)
await ctx.info("Waiting for you to log in and authorize (Browser opened on your computer)...")
server = HTTPServer(("localhost", 8765), CallbackHandler)
# Run handle_request in a separate thread so it doesn't block the async event loop
await asyncio.to_thread(server.handle_request)
if not auth_code:
return [TextContent(type="text", text="Error: No authorization code received.")]
await ctx.info("Authorization code received. Exchanging for tokens...")
async with httpx.AsyncClient() as client:
response = await client.post(
"https://www.strava.com/oauth/token",
data={
"client_id": client_id,
"client_secret": client_secret,
"code": auth_code,
"grant_type": "authorization_code",
},
)
if response.status_code != 200:
return [TextContent(type="text", text=f"Error: Token exchange failed: {response.status_code} {response.text}")]
data = response.json()
refresh_token = data.get("refresh_token")
# Update the .env file if it exists
env_msg = ""
try:
env_path = ".env"
if os.path.exists(env_path):
with open(env_path, "r") as f:
lines = f.readlines()
with open(env_path, "w") as f:
for line in lines:
if line.startswith("STRAVA_REFRESH_TOKEN="):
f.write(f"STRAVA_REFRESH_TOKEN={refresh_token}\n")
else:
f.write(line)
env_msg = "\nI have also automatically updated your .env file with the new refresh token!"
except Exception as e:
env_msg = f"\nFailed to automatically update .env file: {e}"
return [TextContent(type="text", text=f"""
✅ Authorization successful!
You have successfully authenticated with Strava.
Your new Refresh Token is: `{refresh_token}`
{env_msg}
""")]
+65 -6
View File
@@ -1,6 +1,25 @@
import json
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
async def list_athlete_clubs(): async def list_athlete_clubs():
@@ -10,7 +29,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
clubs = await strava.get_athlete_clubs() clubs = await strava.get_athlete_clubs()
return [ data = [
{ {
"id": c.get("id"), "id": c.get("id"),
"name": c.get("name"), "name": c.get("name"),
@@ -22,8 +41,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for c in clubs for c in clubs
] ]
if not data:
md = "### 🏘️ Keine Clubs gefunden."
else:
md = f"### 🏘️ Clubs ({len(data)})\n"
md += "| Name | Sport | Mitglieder | Ort |\n"
md += "|------|-------|------------|-----|\n"
for c in data:
loc = ", ".join(filter(None, [c["city"], c["country"]])) or "N/A"
md += f"| {c['name']} | {c['sport_type'] or 'N/A'} | {c['member_count']} | {loc} |\n"
return [_user_text(md.strip()), _resource("internal://clubs/list", data)]
except Exception as e: except Exception as e:
return f"Error fetching clubs: {str(e)}" return [TextContent(type="text", text=f"Error fetching clubs: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_club(club_id: int): async def get_club(club_id: int):
@@ -45,7 +74,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
activities = await strava.get_club_activities(club_id, per_page=limit) activities = await strava.get_club_activities(club_id, per_page=limit)
return [ data = [
{ {
"name": a.get("name"), "name": a.get("name"),
"sport_type": a.get("sport_type") or a.get("type"), "sport_type": a.get("sport_type") or a.get("type"),
@@ -56,8 +85,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for a in activities for a in activities
] ]
if not data:
md = "### 🚴 Keine Club-Aktivitäten gefunden."
else:
md = f"### 🚴 Club-Aktivitäten ({len(data)})\n"
md += "| Athlet | Sport | Name | Distanz | Zeit |\n"
md += "|--------|-------|------|---------|------|\n"
for a in data:
md += f"| {a['athlete']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://clubs/{club_id}/activities", data),
]
except Exception as e: except Exception as e:
return f"Error fetching club activities: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching club activities: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def get_club_members(club_id: int, limit: int = 30): async def get_club_members(club_id: int, limit: int = 30):
@@ -68,7 +113,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
members = await strava.get_club_members(club_id, per_page=limit) members = await strava.get_club_members(club_id, per_page=limit)
return [ data = [
{ {
"id": m.get("id"), "id": m.get("id"),
"name": f"{m.get('firstname')} {m.get('lastname')}", "name": f"{m.get('firstname')} {m.get('lastname')}",
@@ -77,5 +122,19 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for m in members for m in members
] ]
if not data:
md = "### 👥 Keine Mitglieder gefunden."
else:
md = f"### 👥 Mitglieder ({len(data)})\n"
md += "| Name | Ort |\n|------|-----|\n"
for m in data:
loc = ", ".join(filter(None, [m["city"], m["country"]])) or "N/A"
md += f"| {m['name']} | {loc} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://clubs/{club_id}/members", data),
]
except Exception as e: except Exception as e:
return f"Error fetching club members: {str(e)}" return [
TextContent(type="text", text=f"Error fetching club members: {str(e)}")
]
+40 -2
View File
@@ -1,6 +1,25 @@
import json
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
async def get_gear_by_id(gear_id: str): async def get_gear_by_id(gear_id: str):
@@ -12,7 +31,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
g = await strava.get_gear_by_id(gear_id) g = await strava.get_gear_by_id(gear_id)
return { data = {
"id": g.get("id"), "id": g.get("id"),
"name": g.get("name"), "name": g.get("name"),
"nickname": g.get("nickname"), "nickname": g.get("nickname"),
@@ -24,5 +43,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"primary": g.get("primary", False), "primary": g.get("primary", False),
"retired": g.get("retired", False), "retired": g.get("retired", False),
} }
brand_model = (
" ".join(filter(None, [data["brand_name"], data["model_name"]]))
or "N/A"
)
md = f"""### 🚲 Ausrüstung: {data["name"] or gear_id}
| Feld | Wert |
|------|------|
| Marke / Modell | {brand_model} |
| Spitzname | {data["nickname"] or "N/A"} |
| Typ | {data["type"] or "N/A"} |
| Gesamt-Distanz | {data["distance"]} |
| Primär | {"✅ Ja" if data["primary"] else "Nein"} |
| Im Ruhestand | {"🛑 Ja" if data["retired"] else "Nein"} |"""
if data["description"]:
md += f"\n\n_{data['description']}_"
return [
_user_text(md.strip()),
_resource(f"internal://gear/{gear_id}", data),
]
except Exception as e: except Exception as e:
return f"Error fetching gear: {str(e)}" return [TextContent(type="text", text=f"Error fetching gear: {str(e)}")]
+8 -3
View File
@@ -1,6 +1,7 @@
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.prompt() @mcp.prompt()
def analyze_activity(activity_id: str) -> str: def analyze_activity(activity_id: str) -> str:
@@ -29,11 +30,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
:param weeks: Number of weeks to cover (default 4). :param weeks: Number of weeks to cover (default 4).
Fetches recent activities and athlete stats to produce a summary report. Fetches recent activities and athlete stats to produce a summary report.
""" """
import time from datetime import datetime, timedelta, timezone
after_ts = int(time.time()) - weeks * 7 * 24 * 3600
after_dt = (datetime.now(timezone.utc) - timedelta(weeks=weeks)).replace(
hour=0, minute=0, second=0, microsecond=0
)
after_iso = after_dt.isoformat().replace("+00:00", "Z")
return ( return (
f"Please summarize my Strava training for the last {weeks} weeks.\n\n" f"Please summarize my Strava training for the last {weeks} weeks.\n\n"
f"Use list_activities with after={after_ts} (Unix timestamp = {weeks} weeks ago) " f"Use list_activities with after='{after_iso}' (ISO 8601) "
"and a high limit to fetch all recent activities. " "and a high limit to fetch all recent activities. "
"Also read the get_athlete_stats tool for overall totals.\n\n" "Also read the get_athlete_stats tool for overall totals.\n\n"
"Structure the report as follows:\n" "Structure the report as follows:\n"
+60 -6
View File
@@ -1,6 +1,25 @@
import json
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
async def get_routes_by_athlete_id(limit: int = 30): async def get_routes_by_athlete_id(limit: int = 30):
@@ -11,12 +30,16 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
routes = await strava.get_routes_by_athlete(per_page=min(limit, 200)) routes = await strava.get_routes_by_athlete(per_page=min(limit, 200))
return [ data = [
{ {
"id": str(r.get("id")), "id": str(r.get("id")),
"name": r.get("name"), "name": r.get("name"),
"description": r.get("description") or "", "description": r.get("description") or "",
"type": "Run" if r.get("type") == 1 else "Ride" if r.get("type") == 2 else "Other", "type": "Run"
if r.get("type") == 1
else "Ride"
if r.get("type") == 2
else "Other",
"sub_type": r.get("sub_type"), "sub_type": r.get("sub_type"),
"distance": f"{r.get('distance', 0) / 1000:.2f} km", "distance": f"{r.get('distance', 0) / 1000:.2f} km",
"elevation_gain": f"{r.get('elevation_gain', 0):.0f} m", "elevation_gain": f"{r.get('elevation_gain', 0):.0f} m",
@@ -26,8 +49,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for r in routes for r in routes
] ]
if not data:
md = "### 🗺️ Keine Routen gefunden."
else:
md = f"### 🗺️ Routen ({len(data)})\n"
md += "| Name | Typ | Distanz | Höhenmeter | Dauer |\n"
md += "|------|-----|---------|------------|-------|\n"
for r in data:
star = "" if r["starred"] else ""
md += f"| {star}{r['name']} | {r['type']} | {r['distance']} | {r['elevation_gain']} | {r['estimated_moving_time']} |\n"
return [_user_text(md.strip()), _resource("internal://routes/list", data)]
except Exception as e: except Exception as e:
return f"Error fetching routes: {str(e)}" return [TextContent(type="text", text=f"Error fetching routes: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_route_by_id(route_id: str): async def get_route_by_id(route_id: str):
@@ -38,11 +71,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
r = await strava.get_route_by_id(route_id) r = await strava.get_route_by_id(route_id)
return { data = {
"id": str(r.get("id")), "id": str(r.get("id")),
"name": r.get("name"), "name": r.get("name"),
"description": r.get("description") or "", "description": r.get("description") or "",
"type": "Run" if r.get("type") == 1 else "Ride" if r.get("type") == 2 else "Other", "type": "Run"
if r.get("type") == 1
else "Ride"
if r.get("type") == 2
else "Other",
"distance": f"{r.get('distance', 0) / 1000:.2f} km", "distance": f"{r.get('distance', 0) / 1000:.2f} km",
"elevation_gain": f"{r.get('elevation_gain', 0):.0f} m", "elevation_gain": f"{r.get('elevation_gain', 0):.0f} m",
"estimated_moving_time": f"{r.get('estimated_moving_time', 0) / 60:.0f} min", "estimated_moving_time": f"{r.get('estimated_moving_time', 0) / 60:.0f} min",
@@ -58,8 +95,25 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
for s in r.get("segments", []) for s in r.get("segments", [])
], ],
} }
n_seg = len(data["segments"])
md = f"""### 🗺️ Route: {data["name"]}
| Feld | Wert |
|------|------|
| Typ | {data["type"]} |
| Distanz | {data["distance"]} |
| Höhenmeter | {data["elevation_gain"]} |
| Geschätzte Dauer | {data["estimated_moving_time"]} |
| Segmente | {n_seg} |
| Favorit | {"⭐ Ja" if data["starred"] else "Nein"} |
| Privat | {"🔒 Ja" if data["private"] else "Nein"} |"""
if data["description"]:
md += f"\n\n_{data['description']}_"
return [
_user_text(md.strip()),
_resource(f"internal://routes/{route_id}", data),
]
except Exception as e: except Exception as e:
return f"Error fetching route: {str(e)}" return [TextContent(type="text", text=f"Error fetching route: {str(e)}")]
@mcp.tool() @mcp.tool()
async def get_route_streams(route_id: str): async def get_route_streams(route_id: str):
+70 -6
View File
@@ -1,5 +1,25 @@
import json
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
from strava_mcp_server.utils import format_date_iso, format_date_human
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
@@ -12,12 +32,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
e = await strava.get_segment_effort(effort_id) e = await strava.get_segment_effort(effort_id)
return { data = {
"id": str(e.get("id")), "id": str(e.get("id")),
"name": e.get("name"), "name": e.get("name"),
"elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min", "elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min",
"moving_time": f"{e.get('moving_time', 0) / 60:.1f} min", "moving_time": f"{e.get('moving_time', 0) / 60:.1f} min",
"start_date": e.get("start_date"), "start_date": format_date_iso(e.get("start_date")),
"distance": f"{e.get('distance', 0) / 1000:.2f} km", "distance": f"{e.get('distance', 0) / 1000:.2f} km",
"average_watts": e.get("average_watts"), "average_watts": e.get("average_watts"),
"average_heartrate": e.get("average_heartrate"), "average_heartrate": e.get("average_heartrate"),
@@ -25,8 +45,35 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"pr_rank": e.get("pr_rank"), "pr_rank": e.get("pr_rank"),
"kom_rank": e.get("kom_rank"), "kom_rank": e.get("kom_rank"),
} }
pr = f"#{data['pr_rank']}" if data["pr_rank"] else "N/A"
kom = f"#{data['kom_rank']}" if data["kom_rank"] else "N/A"
w = f"{data['average_watts']:.0f} W" if data["average_watts"] else "N/A"
hr = (
f"{data['average_heartrate']:.0f} bpm"
if data["average_heartrate"]
else "N/A"
)
md = f"""### 🏅 Segment-Effort: {data["name"]}
| Feld | Wert |
|------|------|
| Datum | {format_date_human(data["start_date"])} |
| Distanz | {data["distance"]} |
| Zeit (gesamt) | {data["elapsed_time"]} |
| Fahrzeit | {data["moving_time"]} |
| Ø Leistung | {w} |
| Ø Herzfrequenz | {hr} |
| PR-Rang | {pr} |
| KOM-Rang | {kom} |"""
return [
_user_text(md.strip()),
_resource(f"internal://segment_efforts/{effort_id}", data),
]
except Exception as e: except Exception as e:
return f"Error fetching segment effort: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching segment effort: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def list_segment_efforts( async def list_segment_efforts(
@@ -49,20 +96,37 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
end_date_local=end_date_local, end_date_local=end_date_local,
per_page=limit, per_page=limit,
) )
return [ data = [
{ {
"id": str(e.get("id")), "id": str(e.get("id")),
"elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min", "elapsed_time": f"{e.get('elapsed_time', 0) / 60:.1f} min",
"moving_time": f"{e.get('moving_time', 0) / 60:.1f} min", "moving_time": f"{e.get('moving_time', 0) / 60:.1f} min",
"start_date": e.get("start_date"), "start_date": format_date_iso(e.get("start_date")),
"average_watts": e.get("average_watts"), "average_watts": e.get("average_watts"),
"average_heartrate": e.get("average_heartrate"), "average_heartrate": e.get("average_heartrate"),
"pr_rank": e.get("pr_rank"), "pr_rank": e.get("pr_rank"),
} }
for e in efforts for e in efforts
] ]
if not data:
md = "### 🏅 Keine Efforts für dieses Segment gefunden."
else:
md = f"### 🏅 Segment-Efforts ({len(data)})\n"
md += "| Datum | Zeit | Fahrzeit | PR-Rang |\n"
md += "|-------|------|----------|--------|\n"
for effort in data:
pr = f"#{effort['pr_rank']}" if effort["pr_rank"] else "-"
md += f"| {format_date_human(effort['start_date'])} | {effort['elapsed_time']} | {effort['moving_time']} | {pr} |\n"
return [
_user_text(md.strip()),
_resource(f"internal://segments/{segment_id}/efforts", data),
]
except Exception as e: except Exception as e:
return f"Error fetching segment efforts: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching segment efforts: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def get_segment_effort_streams( async def get_segment_effort_streams(
+75 -6
View File
@@ -1,6 +1,25 @@
import json
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
from strava_mcp_server.strava_client import StravaClient from strava_mcp_server.strava_client import StravaClient
def _resource(uri: str, data) -> EmbeddedResource:
return EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=uri, mimeType="application/json", text=json.dumps(data, indent=2)
),
annotations=Annotations(audience=["assistant"]),
)
def _user_text(text: str) -> TextContent:
return TextContent(
type="text", text=text, annotations=Annotations(audience=["user"])
)
def register(mcp: FastMCP, strava: StravaClient) -> None: def register(mcp: FastMCP, strava: StravaClient) -> None:
@mcp.tool() @mcp.tool()
async def get_segment(segment_id: int): async def get_segment(segment_id: int):
@@ -11,7 +30,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
s = await strava.get_segment(segment_id) s = await strava.get_segment(segment_id)
return { data = {
"id": s.get("id"), "id": s.get("id"),
"name": s.get("name"), "name": s.get("name"),
"activity_type": s.get("activity_type"), "activity_type": s.get("activity_type"),
@@ -28,8 +47,28 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"city": s.get("city"), "city": s.get("city"),
"country": s.get("country"), "country": s.get("country"),
} }
loc = ", ".join(filter(None, [data["city"], data["country"]])) or "N/A"
md = f"""### 📍 Segment: {data["name"]}
| Feld | Wert |
|------|------|
| Sport | {data["activity_type"]} |
| Distanz | {data["distance"]} |
| Ø Steigung | {data["average_grade"]} |
| Max Steigung | {data["maximum_grade"]} |
| Höhe (hoch) | {data["elevation_high"]} |
| Höhe (tief) | {data["elevation_low"]} |
| Höhenmeter | {data["total_elevation_gain"]} |
| Versuche | {data["effort_count"]} |
| Athleten | {data["athlete_count"]} |
| KOM/QOM | {data["kom"] or "N/A"} |
| Ort | {loc} |
| Favorit | {"⭐ Ja" if data["starred"] else "Nein"} |"""
return [
_user_text(md.strip()),
_resource(f"internal://segments/{segment_id}", data),
]
except Exception as e: except Exception as e:
return f"Error fetching segment: {str(e)}" return [TextContent(type="text", text=f"Error fetching segment: {str(e)}")]
@mcp.tool() @mcp.tool()
async def list_starred_segments(limit: int = 30): async def list_starred_segments(limit: int = 30):
@@ -39,7 +78,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
segments = await strava.get_starred_segments(per_page=limit) segments = await strava.get_starred_segments(per_page=limit)
return [ data = [
{ {
"id": s.get("id"), "id": s.get("id"),
"name": s.get("name"), "name": s.get("name"),
@@ -51,8 +90,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for s in segments for s in segments
] ]
if not data:
md = "### ⭐ Keine favorisierten Segmente."
else:
md = f"### ⭐ Favorisierte Segmente ({len(data)})\n"
md += "| Name | Sport | Distanz | Ø Steigung | Versuche |\n"
md += "|------|-------|---------|------------|----------|\n"
for s in data:
md += f"| {s['name']} | {s['activity_type']} | {s['distance']} | {s['average_grade']} | {s['effort_count']} |\n"
return [
_user_text(md.strip()),
_resource("internal://segments/starred", data),
]
except Exception as e: except Exception as e:
return f"Error fetching starred segments: {str(e)}" return [
TextContent(
type="text", text=f"Error fetching starred segments: {str(e)}"
)
]
@mcp.tool() @mcp.tool()
async def explore_segments( async def explore_segments(
@@ -68,7 +123,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
""" """
try: try:
result = await strava.explore_segments(bounds, activity_type) result = await strava.explore_segments(bounds, activity_type)
return [ data = [
{ {
"id": s.get("id"), "id": s.get("id"),
"name": s.get("name"), "name": s.get("name"),
@@ -79,8 +134,22 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
} }
for s in result.get("segments", []) for s in result.get("segments", [])
] ]
if not data:
md = "### 🗺️ Keine Segmente in diesem Bereich gefunden."
else:
md = f"### 🗺️ Segmente in der Region ({len(data)})\n"
md += "| Name | Distanz | Ø Steigung | Höhendiff | Kategorie |\n"
md += "|------|---------|------------|-----------|----------|\n"
for s in data:
md += f"| {s['name']} | {s['distance']} | {s['average_grade']} | {s['elevation_difference']} | {s['climb_category']} |\n"
return [
_user_text(md.strip()),
_resource("internal://segments/explore", data),
]
except Exception as e: except Exception as e:
return f"Error exploring segments: {str(e)}" return [
TextContent(type="text", text=f"Error exploring segments: {str(e)}")
]
@mcp.tool() @mcp.tool()
async def get_segment_streams( async def get_segment_streams(
+63
View File
@@ -0,0 +1,63 @@
from datetime import datetime, timezone
from typing import Optional
def parse_iso_to_unix(iso_str: Optional[str]) -> Optional[int]:
"""
Parses an ISO 8601 string into a Unix timestamp.
Accepts formats like '2024-01-01', '2024-01-01T12:00:00Z', etc.
"""
if not iso_str:
return None
try:
# Remove 'Z' and replace with +00:00 for fromisoformat compatibility if needed
# but fromisoformat in Python 3.11+ handles Z correctly.
# For older versions or varied formats, we use a slightly more robust approach.
clean_iso = iso_str.replace("Z", "+00:00")
dt = datetime.fromisoformat(clean_iso)
# Ensure it has a timezone; default to UTC if missing
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return int(dt.timestamp())
except Exception:
return None
def format_date_iso(date_input: Optional[str | datetime]) -> str:
"""
Standardizes a date string or datetime object to ISO 8601 (UTC).
"""
if not date_input:
return "N/A"
try:
if isinstance(date_input, str):
# Strava dates are often '2024-01-01T12:00:00Z'
dt = datetime.fromisoformat(date_input.replace("Z", "+00:00"))
else:
dt = date_input
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat().replace("+00:00", "Z")
except Exception:
return str(date_input)
def format_date_human(date_input: Optional[str | datetime]) -> str:
"""
Returns a human-readable date/time string (DD.MM.YYYY HH:MM).
"""
if not date_input:
return "N/A"
try:
if isinstance(date_input, str):
dt = datetime.fromisoformat(date_input.replace("Z", "+00:00"))
else:
dt = date_input
return dt.strftime("%d.%m.%Y %H:%M")
except Exception:
return str(date_input)
+11 -3
View File
@@ -3,6 +3,7 @@ import os
from strava_client import StravaClient from strava_client import StravaClient
from dotenv import load_dotenv from dotenv import load_dotenv
async def test_connection(): async def test_connection():
load_dotenv() load_dotenv()
@@ -12,7 +13,9 @@ async def test_connection():
if not all([client_id, client_secret, refresh_token]): if not all([client_id, client_secret, refresh_token]):
print("❌ Error: Missing Strava credentials in .env file.") print("❌ Error: Missing Strava credentials in .env file.")
print("Please ensure STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, and STRAVA_REFRESH_TOKEN are set.") print(
"Please ensure STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, and STRAVA_REFRESH_TOKEN are set."
)
return return
client = StravaClient() client = StravaClient()
@@ -20,7 +23,9 @@ async def test_connection():
print("Testing Strava connection...") print("Testing Strava connection...")
try: try:
athlete = await client.get_athlete() athlete = await client.get_athlete()
print(f"✅ Success! Connected as {athlete.get('firstname')} {athlete.get('lastname')}") print(
f"✅ Success! Connected as {athlete.get('firstname')} {athlete.get('lastname')}"
)
print(f"Athlete ID: {athlete.get('id')}") print(f"Athlete ID: {athlete.get('id')}")
print("\nFetching recent activities...") print("\nFetching recent activities...")
@@ -32,8 +37,11 @@ async def test_connection():
except Exception as e: except Exception as e:
print(f"❌ Connection failed: {str(e)}") print(f"❌ Connection failed: {str(e)}")
if "401" in str(e): if "401" in str(e):
print("Hint: This is likely a scope issue. Your token needs 'activity:read' permission.") print(
"Hint: This is likely a scope issue. Your token needs 'activity:read' permission."
)
print("Run: uv run get_token.py to re-authorize with the correct scopes.") print("Run: uv run get_token.py to re-authorize with the correct scopes.")
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(test_connection()) asyncio.run(test_connection())
+1
View File
@@ -0,0 +1 @@
# This file marks the tests/unit directory as a Python package.
+63
View File
@@ -0,0 +1,63 @@
"""Unit tests for MCP content block helper functions."""
import json
from mcp.types import TextContent, EmbeddedResource
# Import helpers directly from a tool module to test them
from strava_mcp_server.tools.activities import _resource, _user_text
class TestUserTextHelper:
def test_returns_text_content(self):
result = _user_text("Hello World")
assert isinstance(result, TextContent)
def test_type_is_text(self):
result = _user_text("Hello")
assert result.type == "text"
def test_text_value(self):
result = _user_text("Test message")
assert result.text == "Test message"
def test_audience_is_user(self):
result = _user_text("Hello")
assert result.annotations is not None
assert result.annotations.audience == ["user"]
class TestResourceHelper:
def test_returns_embedded_resource(self):
result = _resource("internal://test/data", {"key": "value"})
assert isinstance(result, EmbeddedResource)
def test_type_is_resource(self):
result = _resource("internal://test/data", {"key": "value"})
assert result.type == "resource"
def test_mime_type_is_json(self):
result = _resource("internal://test/data", {"key": "value"})
assert result.resource.mimeType == "application/json"
def test_uri_is_set(self):
result = _resource("internal://athlete/profile", {"id": 1})
assert str(result.resource.uri) == "internal://athlete/profile"
def test_text_is_valid_json(self):
data = {"id": 42, "name": "Test Athlete", "active": True}
result = _resource("internal://test", data)
parsed = json.loads(result.resource.text)
assert parsed == data
def test_audience_is_assistant(self):
result = _resource("internal://test", {})
assert result.annotations is not None
assert result.annotations.audience == ["assistant"]
def test_list_data(self):
data = [{"id": 1}, {"id": 2}]
result = _resource("internal://activities/list", data)
parsed = json.loads(result.resource.text)
assert len(parsed) == 2
assert parsed[0]["id"] == 1
+98
View File
@@ -0,0 +1,98 @@
"""Unit tests for strava_mcp_server.utils — pure functions, no external dependencies."""
from datetime import datetime, timezone
from strava_mcp_server.utils import (
parse_iso_to_unix,
format_date_iso,
format_date_human,
)
class TestParseIsoToUnix:
def test_full_iso_with_z(self):
result = parse_iso_to_unix("2024-01-15T12:00:00Z")
assert result == int(
datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc).timestamp()
)
def test_full_iso_with_offset(self):
result = parse_iso_to_unix("2024-01-15T13:00:00+01:00")
assert result == int(
datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc).timestamp()
)
def test_date_only(self):
result = parse_iso_to_unix("2024-06-01")
assert result == int(
datetime(2024, 6, 1, 0, 0, 0, tzinfo=timezone.utc).timestamp()
)
def test_none_returns_none(self):
assert parse_iso_to_unix(None) is None
def test_empty_string_returns_none(self):
assert parse_iso_to_unix("") is None
def test_invalid_string_returns_none(self):
assert parse_iso_to_unix("not-a-date") is None
def test_returns_int(self):
result = parse_iso_to_unix("2024-01-01T00:00:00Z")
assert isinstance(result, int)
class TestFormatDateIso:
def test_z_suffix_normalized(self):
result = format_date_iso("2024-01-15T12:00:00Z")
assert result == "2024-01-15T12:00:00Z"
def test_offset_preserved(self):
# format_date_iso preserves timezone offset info; normalization to UTC
# is the caller's responsibility if needed.
result = format_date_iso("2024-01-15T13:00:00+01:00")
assert result == "2024-01-15T13:00:00+01:00"
def test_datetime_object(self):
dt = datetime(2024, 3, 10, 8, 30, 0, tzinfo=timezone.utc)
result = format_date_iso(dt)
assert result == "2024-03-10T08:30:00Z"
def test_none_returns_na(self):
assert format_date_iso(None) == "N/A"
def test_empty_string_returns_na(self):
assert format_date_iso("") == "N/A"
def test_invalid_string_returned_as_is(self):
result = format_date_iso("not-a-date")
assert result == "not-a-date"
class TestFormatDateHuman:
def test_iso_with_z(self):
result = format_date_human("2024-01-15T08:30:00Z")
assert result == "15.01.2024 08:30"
def test_iso_with_offset(self):
# +01:00 → displayed in local time of the datetime (13:00 in +01 = 13:00 displayed)
result = format_date_human("2024-01-15T13:45:00+01:00")
assert result == "15.01.2024 13:45"
def test_datetime_object(self):
dt = datetime(2024, 12, 31, 23, 59, tzinfo=timezone.utc)
result = format_date_human(dt)
assert result == "31.12.2024 23:59"
def test_none_returns_na(self):
assert format_date_human(None) == "N/A"
def test_empty_string_returns_na(self):
assert format_date_human("") == "N/A"
def test_format_pattern(self):
result = format_date_human("2024-06-01T00:00:00Z")
# Ensure it matches DD.MM.YYYY HH:MM
import re
assert re.match(r"\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}", result)
Generated
+153 -1
View File
@@ -48,6 +48,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" }, { url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
] ]
[[package]]
name = "backports-asyncio-runner"
version = "1.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8e/ff/70dca7d7cb1cbc0edb2c6cc0c38b65cba36cccc491eca64cabd5fe7f8670/backports_asyncio_runner-1.2.0.tar.gz", hash = "sha256:a5aa7b2b7d8f8bfcaa2b57313f70792df84e32a2a746f585213373f900b42162", size = 69893, upload-time = "2025-07-02T02:27:15.685Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a0/59/76ab57e3fe74484f48a53f8e337171b4a2349e506eabe136d7e01d059086/backports_asyncio_runner-1.2.0-py3-none-any.whl", hash = "sha256:0da0a936a8aeb554eccb426dc55af3ba63bcdc69fa1a600b5bb305413a4477b5", size = 12313, upload-time = "2025-07-02T02:27:14.263Z" },
]
[[package]] [[package]]
name = "certifi" name = "certifi"
version = "2026.4.22" version = "2026.4.22"
@@ -303,6 +312,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/5d/13/ad7d7ca3808a898b4612b6fe93cde56b53f3034dcde235acb1f0e1df24c6/idna-3.13-py3-none-any.whl", hash = "sha256:892ea0cde124a99ce773decba204c5552b69c3c67ffd5f232eb7696135bc8bb3", size = 68629, upload-time = "2026-04-22T16:42:40.909Z" }, { url = "https://files.pythonhosted.org/packages/5d/13/ad7d7ca3808a898b4612b6fe93cde56b53f3034dcde235acb1f0e1df24c6/idna-3.13-py3-none-any.whl", hash = "sha256:892ea0cde124a99ce773decba204c5552b69c3c67ffd5f232eb7696135bc8bb3", size = 68629, upload-time = "2026-04-22T16:42:40.909Z" },
] ]
[[package]]
name = "iniconfig"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
[[package]] [[package]]
name = "jsonschema" name = "jsonschema"
version = "4.26.0" version = "4.26.0"
@@ -382,6 +400,24 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
] ]
[[package]]
name = "packaging"
version = "26.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]] [[package]]
name = "pycparser" name = "pycparser"
version = "3.0" version = "3.0"
@@ -562,6 +598,51 @@ crypto = [
{ name = "cryptography" }, { name = "cryptography" },
] ]
[[package]]
name = "pytest"
version = "9.0.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "exceptiongroup", marker = "python_full_version < '3.11'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
{ name = "tomli", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165, upload-time = "2026-04-07T17:16:18.027Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" },
]
[[package]]
name = "pytest-asyncio"
version = "1.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" },
{ name = "pytest" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
]
[[package]]
name = "pytest-sugar"
version = "1.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
{ name = "termcolor" },
]
sdist = { url = "https://files.pythonhosted.org/packages/0b/4e/60fed105549297ba1a700e1ea7b828044842ea27d72c898990510b79b0e2/pytest-sugar-1.1.1.tar.gz", hash = "sha256:73b8b65163ebf10f9f671efab9eed3d56f20d2ca68bda83fa64740a92c08f65d", size = 16533, upload-time = "2025-08-23T12:19:35.737Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/87/d5/81d38a91c1fdafb6711f053f5a9b92ff788013b19821257c2c38c1e132df/pytest_sugar-1.1.1-py3-none-any.whl", hash = "sha256:2f8319b907548d5b9d03a171515c1d43d2e38e32bd8182a1781eb20b43344cc8", size = 11440, upload-time = "2025-08-23T12:19:34.894Z" },
]
[[package]] [[package]]
name = "python-dotenv" name = "python-dotenv"
version = "1.2.2" version = "1.2.2"
@@ -824,6 +905,9 @@ dependencies = [
[package.dev-dependencies] [package.dev-dependencies]
dev = [ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-sugar" },
{ name = "ruff" }, { name = "ruff" },
] ]
@@ -837,7 +921,75 @@ requires-dist = [
] ]
[package.metadata.requires-dev] [package.metadata.requires-dev]
dev = [{ name = "ruff", specifier = ">=0.15.12" }] dev = [
{ name = "pytest", specifier = ">=8.0" },
{ name = "pytest-asyncio", specifier = ">=0.24" },
{ name = "pytest-sugar", specifier = ">=1.1.1" },
{ name = "ruff", specifier = ">=0.15.12" },
]
[[package]]
name = "termcolor"
version = "3.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/46/79/cf31d7a93a8fdc6aa0fbb665be84426a8c5a557d9240b6239e9e11e35fc5/termcolor-3.3.0.tar.gz", hash = "sha256:348871ca648ec6a9a983a13ab626c0acce02f515b9e1983332b17af7979521c5", size = 14434, upload-time = "2025-12-29T12:55:21.882Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/33/d1/8bb87d21e9aeb323cc03034f5eaf2c8f69841e40e4853c2627edf8111ed3/termcolor-3.3.0-py3-none-any.whl", hash = "sha256:cf642efadaf0a8ebbbf4bc7a31cec2f9b5f21a9f726f4ccbb08192c9c26f43a5", size = 7734, upload-time = "2025-12-29T12:55:20.718Z" },
]
[[package]]
name = "tomli"
version = "2.4.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/22/de/48c59722572767841493b26183a0d1cc411d54fd759c5607c4590b6563a6/tomli-2.4.1.tar.gz", hash = "sha256:7c7e1a961a0b2f2472c1ac5b69affa0ae1132c39adcb67aba98568702b9cc23f", size = 17543, upload-time = "2026-03-25T20:22:03.828Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/f4/11/db3d5885d8528263d8adc260bb2d28ebf1270b96e98f0e0268d32b8d9900/tomli-2.4.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:f8f0fc26ec2cc2b965b7a3b87cd19c5c6b8c5e5f436b984e85f486d652285c30", size = 154704, upload-time = "2026-03-25T20:21:10.473Z" },
{ url = "https://files.pythonhosted.org/packages/6d/f7/675db52c7e46064a9aa928885a9b20f4124ecb9bc2e1ce74c9106648d202/tomli-2.4.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4ab97e64ccda8756376892c53a72bd1f964e519c77236368527f758fbc36a53a", size = 149454, upload-time = "2026-03-25T20:21:12.036Z" },
{ url = "https://files.pythonhosted.org/packages/61/71/81c50943cf953efa35bce7646caab3cf457a7d8c030b27cfb40d7235f9ee/tomli-2.4.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:96481a5786729fd470164b47cdb3e0e58062a496f455ee41b4403be77cb5a076", size = 237561, upload-time = "2026-03-25T20:21:13.098Z" },
{ url = "https://files.pythonhosted.org/packages/48/c1/f41d9cb618acccca7df82aaf682f9b49013c9397212cb9f53219e3abac37/tomli-2.4.1-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5a881ab208c0baf688221f8cecc5401bd291d67e38a1ac884d6736cbcd8247e9", size = 243824, upload-time = "2026-03-25T20:21:14.569Z" },
{ url = "https://files.pythonhosted.org/packages/22/e4/5a816ecdd1f8ca51fb756ef684b90f2780afc52fc67f987e3c61d800a46d/tomli-2.4.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:47149d5bd38761ac8be13a84864bf0b7b70bc051806bc3669ab1cbc56216b23c", size = 242227, upload-time = "2026-03-25T20:21:15.712Z" },
{ url = "https://files.pythonhosted.org/packages/6b/49/2b2a0ef529aa6eec245d25f0c703e020a73955ad7edf73e7f54ddc608aa5/tomli-2.4.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:ec9bfaf3ad2df51ace80688143a6a4ebc09a248f6ff781a9945e51937008fcbc", size = 247859, upload-time = "2026-03-25T20:21:17.001Z" },
{ url = "https://files.pythonhosted.org/packages/83/bd/6c1a630eaca337e1e78c5903104f831bda934c426f9231429396ce3c3467/tomli-2.4.1-cp311-cp311-win32.whl", hash = "sha256:ff2983983d34813c1aeb0fa89091e76c3a22889ee83ab27c5eeb45100560c049", size = 97204, upload-time = "2026-03-25T20:21:18.079Z" },
{ url = "https://files.pythonhosted.org/packages/42/59/71461df1a885647e10b6bb7802d0b8e66480c61f3f43079e0dcd315b3954/tomli-2.4.1-cp311-cp311-win_amd64.whl", hash = "sha256:5ee18d9ebdb417e384b58fe414e8d6af9f4e7a0ae761519fb50f721de398dd4e", size = 108084, upload-time = "2026-03-25T20:21:18.978Z" },
{ url = "https://files.pythonhosted.org/packages/b8/83/dceca96142499c069475b790e7913b1044c1a4337e700751f48ed723f883/tomli-2.4.1-cp311-cp311-win_arm64.whl", hash = "sha256:c2541745709bad0264b7d4705ad453b76ccd191e64aa6f0fc66b69a293a45ece", size = 95285, upload-time = "2026-03-25T20:21:20.309Z" },
{ url = "https://files.pythonhosted.org/packages/c1/ba/42f134a3fe2b370f555f44b1d72feebb94debcab01676bf918d0cb70e9aa/tomli-2.4.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:c742f741d58a28940ce01d58f0ab2ea3ced8b12402f162f4d534dfe18ba1cd6a", size = 155924, upload-time = "2026-03-25T20:21:21.626Z" },
{ url = "https://files.pythonhosted.org/packages/dc/c7/62d7a17c26487ade21c5422b646110f2162f1fcc95980ef7f63e73c68f14/tomli-2.4.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7f86fd587c4ed9dd76f318225e7d9b29cfc5a9d43de44e5754db8d1128487085", size = 150018, upload-time = "2026-03-25T20:21:23.002Z" },
{ url = "https://files.pythonhosted.org/packages/5c/05/79d13d7c15f13bdef410bdd49a6485b1c37d28968314eabee452c22a7fda/tomli-2.4.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ff18e6a727ee0ab0388507b89d1bc6a22b138d1e2fa56d1ad494586d61d2eae9", size = 244948, upload-time = "2026-03-25T20:21:24.04Z" },
{ url = "https://files.pythonhosted.org/packages/10/90/d62ce007a1c80d0b2c93e02cab211224756240884751b94ca72df8a875ca/tomli-2.4.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:136443dbd7e1dee43c68ac2694fde36b2849865fa258d39bf822c10e8068eac5", size = 253341, upload-time = "2026-03-25T20:21:25.177Z" },
{ url = "https://files.pythonhosted.org/packages/1a/7e/caf6496d60152ad4ed09282c1885cca4eea150bfd007da84aea07bcc0a3e/tomli-2.4.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:5e262d41726bc187e69af7825504c933b6794dc3fbd5945e41a79bb14c31f585", size = 248159, upload-time = "2026-03-25T20:21:26.364Z" },
{ url = "https://files.pythonhosted.org/packages/99/e7/c6f69c3120de34bbd882c6fba7975f3d7a746e9218e56ab46a1bc4b42552/tomli-2.4.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5cb41aa38891e073ee49d55fbc7839cfdb2bc0e600add13874d048c94aadddd1", size = 253290, upload-time = "2026-03-25T20:21:27.46Z" },
{ url = "https://files.pythonhosted.org/packages/d6/2f/4a3c322f22c5c66c4b836ec58211641a4067364f5dcdd7b974b4c5da300c/tomli-2.4.1-cp312-cp312-win32.whl", hash = "sha256:da25dc3563bff5965356133435b757a795a17b17d01dbc0f42fb32447ddfd917", size = 98141, upload-time = "2026-03-25T20:21:28.492Z" },
{ url = "https://files.pythonhosted.org/packages/24/22/4daacd05391b92c55759d55eaee21e1dfaea86ce5c571f10083360adf534/tomli-2.4.1-cp312-cp312-win_amd64.whl", hash = "sha256:52c8ef851d9a240f11a88c003eacb03c31fc1c9c4ec64a99a0f922b93874fda9", size = 108847, upload-time = "2026-03-25T20:21:29.386Z" },
{ url = "https://files.pythonhosted.org/packages/68/fd/70e768887666ddd9e9f5d85129e84910f2db2796f9096aa02b721a53098d/tomli-2.4.1-cp312-cp312-win_arm64.whl", hash = "sha256:f758f1b9299d059cc3f6546ae2af89670cb1c4d48ea29c3cacc4fe7de3058257", size = 95088, upload-time = "2026-03-25T20:21:30.677Z" },
{ url = "https://files.pythonhosted.org/packages/07/06/b823a7e818c756d9a7123ba2cda7d07bc2dd32835648d1a7b7b7a05d848d/tomli-2.4.1-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:36d2bd2ad5fb9eaddba5226aa02c8ec3fa4f192631e347b3ed28186d43be6b54", size = 155866, upload-time = "2026-03-25T20:21:31.65Z" },
{ url = "https://files.pythonhosted.org/packages/14/6f/12645cf7f08e1a20c7eb8c297c6f11d31c1b50f316a7e7e1e1de6e2e7b7e/tomli-2.4.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:eb0dc4e38e6a1fd579e5d50369aa2e10acfc9cace504579b2faabb478e76941a", size = 149887, upload-time = "2026-03-25T20:21:33.028Z" },
{ url = "https://files.pythonhosted.org/packages/5c/e0/90637574e5e7212c09099c67ad349b04ec4d6020324539297b634a0192b0/tomli-2.4.1-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c7f2c7f2b9ca6bdeef8f0fa897f8e05085923eb091721675170254cbc5b02897", size = 243704, upload-time = "2026-03-25T20:21:34.51Z" },
{ url = "https://files.pythonhosted.org/packages/10/8f/d3ddb16c5a4befdf31a23307f72828686ab2096f068eaf56631e136c1fdd/tomli-2.4.1-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f3c6818a1a86dd6dca7ddcaaf76947d5ba31aecc28cb1b67009a5877c9a64f3f", size = 251628, upload-time = "2026-03-25T20:21:36.012Z" },
{ url = "https://files.pythonhosted.org/packages/e3/f1/dbeeb9116715abee2485bf0a12d07a8f31af94d71608c171c45f64c0469d/tomli-2.4.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:d312ef37c91508b0ab2cee7da26ec0b3ed2f03ce12bd87a588d771ae15dcf82d", size = 247180, upload-time = "2026-03-25T20:21:37.136Z" },
{ url = "https://files.pythonhosted.org/packages/d3/74/16336ffd19ed4da28a70959f92f506233bd7cfc2332b20bdb01591e8b1d1/tomli-2.4.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:51529d40e3ca50046d7606fa99ce3956a617f9b36380da3b7f0dd3dd28e68cb5", size = 251674, upload-time = "2026-03-25T20:21:38.298Z" },
{ url = "https://files.pythonhosted.org/packages/16/f9/229fa3434c590ddf6c0aa9af64d3af4b752540686cace29e6281e3458469/tomli-2.4.1-cp313-cp313-win32.whl", hash = "sha256:2190f2e9dd7508d2a90ded5ed369255980a1bcdd58e52f7fe24b8162bf9fedbd", size = 97976, upload-time = "2026-03-25T20:21:39.316Z" },
{ url = "https://files.pythonhosted.org/packages/6a/1e/71dfd96bcc1c775420cb8befe7a9d35f2e5b1309798f009dca17b7708c1e/tomli-2.4.1-cp313-cp313-win_amd64.whl", hash = "sha256:8d65a2fbf9d2f8352685bc1364177ee3923d6baf5e7f43ea4959d7d8bc326a36", size = 108755, upload-time = "2026-03-25T20:21:40.248Z" },
{ url = "https://files.pythonhosted.org/packages/83/7a/d34f422a021d62420b78f5c538e5b102f62bea616d1d75a13f0a88acb04a/tomli-2.4.1-cp313-cp313-win_arm64.whl", hash = "sha256:4b605484e43cdc43f0954ddae319fb75f04cc10dd80d830540060ee7cd0243cd", size = 95265, upload-time = "2026-03-25T20:21:41.219Z" },
{ url = "https://files.pythonhosted.org/packages/3c/fb/9a5c8d27dbab540869f7c1f8eb0abb3244189ce780ba9cd73f3770662072/tomli-2.4.1-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:fd0409a3653af6c147209d267a0e4243f0ae46b011aa978b1080359fddc9b6cf", size = 155726, upload-time = "2026-03-25T20:21:42.23Z" },
{ url = "https://files.pythonhosted.org/packages/62/05/d2f816630cc771ad836af54f5001f47a6f611d2d39535364f148b6a92d6b/tomli-2.4.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:a120733b01c45e9a0c34aeef92bf0cf1d56cfe81ed9d47d562f9ed591a9828ac", size = 149859, upload-time = "2026-03-25T20:21:43.386Z" },
{ url = "https://files.pythonhosted.org/packages/ce/48/66341bdb858ad9bd0ceab5a86f90eddab127cf8b046418009f2125630ecb/tomli-2.4.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:559db847dc486944896521f68d8190be1c9e719fced785720d2216fe7022b662", size = 244713, upload-time = "2026-03-25T20:21:44.474Z" },
{ url = "https://files.pythonhosted.org/packages/df/6d/c5fad00d82b3c7a3ab6189bd4b10e60466f22cfe8a08a9394185c8a8111c/tomli-2.4.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:01f520d4f53ef97964a240a035ec2a869fe1a37dde002b57ebc4417a27ccd853", size = 252084, upload-time = "2026-03-25T20:21:45.62Z" },
{ url = "https://files.pythonhosted.org/packages/00/71/3a69e86f3eafe8c7a59d008d245888051005bd657760e96d5fbfb0b740c2/tomli-2.4.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7f94b27a62cfad8496c8d2513e1a222dd446f095fca8987fceef261225538a15", size = 247973, upload-time = "2026-03-25T20:21:46.937Z" },
{ url = "https://files.pythonhosted.org/packages/67/50/361e986652847fec4bd5e4a0208752fbe64689c603c7ae5ea7cb16b1c0ca/tomli-2.4.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ede3e6487c5ef5d28634ba3f31f989030ad6af71edfb0055cbbd14189ff240ba", size = 256223, upload-time = "2026-03-25T20:21:48.467Z" },
{ url = "https://files.pythonhosted.org/packages/8c/9a/b4173689a9203472e5467217e0154b00e260621caa227b6fa01feab16998/tomli-2.4.1-cp314-cp314-win32.whl", hash = "sha256:3d48a93ee1c9b79c04bb38772ee1b64dcf18ff43085896ea460ca8dec96f35f6", size = 98973, upload-time = "2026-03-25T20:21:49.526Z" },
{ url = "https://files.pythonhosted.org/packages/14/58/640ac93bf230cd27d002462c9af0d837779f8773bc03dee06b5835208214/tomli-2.4.1-cp314-cp314-win_amd64.whl", hash = "sha256:88dceee75c2c63af144e456745e10101eb67361050196b0b6af5d717254dddf7", size = 109082, upload-time = "2026-03-25T20:21:50.506Z" },
{ url = "https://files.pythonhosted.org/packages/d5/2f/702d5e05b227401c1068f0d386d79a589bb12bf64c3d2c72ce0631e3bc49/tomli-2.4.1-cp314-cp314-win_arm64.whl", hash = "sha256:b8c198f8c1805dc42708689ed6864951fd2494f924149d3e4bce7710f8eb5232", size = 96490, upload-time = "2026-03-25T20:21:51.474Z" },
{ url = "https://files.pythonhosted.org/packages/45/4b/b877b05c8ba62927d9865dd980e34a755de541eb65fffba52b4cc495d4d2/tomli-2.4.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:d4d8fe59808a54658fcc0160ecfb1b30f9089906c50b23bcb4c69eddc19ec2b4", size = 164263, upload-time = "2026-03-25T20:21:52.543Z" },
{ url = "https://files.pythonhosted.org/packages/24/79/6ab420d37a270b89f7195dec5448f79400d9e9c1826df982f3f8e97b24fd/tomli-2.4.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7008df2e7655c495dd12d2a4ad038ff878d4ca4b81fccaf82b714e07eae4402c", size = 160736, upload-time = "2026-03-25T20:21:53.674Z" },
{ url = "https://files.pythonhosted.org/packages/02/e0/3630057d8eb170310785723ed5adcdfb7d50cb7e6455f85ba8a3deed642b/tomli-2.4.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1d8591993e228b0c930c4bb0db464bdad97b3289fb981255d6c9a41aedc84b2d", size = 270717, upload-time = "2026-03-25T20:21:55.129Z" },
{ url = "https://files.pythonhosted.org/packages/7a/b4/1613716072e544d1a7891f548d8f9ec6ce2faf42ca65acae01d76ea06bb0/tomli-2.4.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:734e20b57ba95624ecf1841e72b53f6e186355e216e5412de414e3c51e5e3c41", size = 278461, upload-time = "2026-03-25T20:21:56.228Z" },
{ url = "https://files.pythonhosted.org/packages/05/38/30f541baf6a3f6df77b3df16b01ba319221389e2da59427e221ef417ac0c/tomli-2.4.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:8a650c2dbafa08d42e51ba0b62740dae4ecb9338eefa093aa5c78ceb546fcd5c", size = 274855, upload-time = "2026-03-25T20:21:57.653Z" },
{ url = "https://files.pythonhosted.org/packages/77/a3/ec9dd4fd2c38e98de34223b995a3b34813e6bdadf86c75314c928350ed14/tomli-2.4.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:504aa796fe0569bb43171066009ead363de03675276d2d121ac1a4572397870f", size = 283144, upload-time = "2026-03-25T20:21:59.089Z" },
{ url = "https://files.pythonhosted.org/packages/ef/be/605a6261cac79fba2ec0c9827e986e00323a1945700969b8ee0b30d85453/tomli-2.4.1-cp314-cp314t-win32.whl", hash = "sha256:b1d22e6e9387bf4739fbe23bfa80e93f6b0373a7f1b96c6227c32bef95a4d7a8", size = 108683, upload-time = "2026-03-25T20:22:00.214Z" },
{ url = "https://files.pythonhosted.org/packages/12/64/da524626d3b9cc40c168a13da8335fe1c51be12c0a63685cc6db7308daae/tomli-2.4.1-cp314-cp314t-win_amd64.whl", hash = "sha256:2c1c351919aca02858f740c6d33adea0c5deea37f9ecca1cc1ef9e884a619d26", size = 121196, upload-time = "2026-03-25T20:22:01.169Z" },
{ url = "https://files.pythonhosted.org/packages/5a/cd/e80b62269fc78fc36c9af5a6b89c835baa8af28ff5ad28c7028d60860320/tomli-2.4.1-cp314-cp314t-win_arm64.whl", hash = "sha256:eab21f45c7f66c13f2a9e0e1535309cee140182a9cdae1e041d02e47291e8396", size = 100393, upload-time = "2026-03-25T20:22:02.137Z" },
{ url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" },
]
[[package]] [[package]]
name = "typer" name = "typer"