10 Commits

Author SHA1 Message Date
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
matthias 4489e1e0e2 refactor: remove unnecessary package write permissions from CI workflow
CI/CD Pipeline / Lint & Check (push) Successful in 9s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 1m17s
2026-05-09 14:53:08 +02:00
matthias 3ce6540a8f chore: update CI/CD runner to ubuntu-latest
CI/CD Pipeline / Lint & Check (push) Successful in 1m23s
CI/CD Pipeline / Build & Push Docker Image (push) Successful in 4m35s
2026-05-09 14:03:13 +02:00
19 changed files with 779 additions and 392 deletions
+3 -7
View File
@@ -38,10 +38,7 @@ jobs:
name: Build & Push Docker Image
needs: lint
if: github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v')
runs-on: gitea-runner-on-dsm
permissions:
packages: write
contents: read
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
@@ -106,9 +103,8 @@ jobs:
# Check if Docker Image section already exists
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 body "$NEW_BODY" '{body: $body}' | \
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```")}' | \
curl -s -X PATCH \
-H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
-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
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
# 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"]
+87 -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)
- 💬 **2 MCP Prompts** for structured AI-driven training analysis
- 🔄 **Fully Automated OAuth** — authentication flow integrated directly as an MCP tool with auto-rotation
- 🐳 **Docker-Ready** highly optimized multi-stage Docker build utilizing `uv`
- 🔄 **Automated OAuth** — authentication flow via a standalone setup script with auto-rotation
- 🐳 **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)
- 🔒 **Read-only** — no write operations, safe to use with AI agents
- 📝 **Design Decisions** — documented architectural choices in `docs/DESIGN_DECISIONS.md`
---
@@ -24,9 +27,10 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
- [Docker (Recommended)](#docker-recommended)
- [Local Python (uv)](#local-python-uv)
- [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)
- [Project Structure](#project-structure)
- [Design Decisions](#design-decisions)
- [CI/CD (Gitea Actions)](#cicd-gitea-actions)
- [Known Strava API Limitations](#known-strava-api-limitations)
- [Troubleshooting](#troubleshooting)
@@ -37,7 +41,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 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 +49,14 @@ Built with [FastMCP](https://github.com/jlowin/fastmcp) and the [MCP Python SDK]
### Docker (Recommended)
The project includes a highly optimized, deterministic Dockerfile powered by `uv`.
The project includes a multi-arch Docker build (amd64/arm64).
```bash
# Clone the repository
git clone https://git.hnrx.net/hnrx/strava-mcp-server.git
cd strava-mcp-server
# Build the image
# Build the image locally
docker build -t strava-mcp-server:latest .
# Run the container (injecting your .env file)
@@ -61,24 +65,29 @@ docker run --rm -p 8000:8000 --env-file .env strava-mcp-server:latest
### Local Python (uv)
We use `uv` for lightning-fast dependency management and task execution.
```bash
git clone https://git.hnrx.net/hnrx/strava-mcp-server.git
cd strava-mcp-server
# Install dependencies and start the server
uv run strava-mcp
# Start the MCP server
uv run server
# Run the OAuth setup script
uv run auth
```
### 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
# 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 +114,22 @@ STRAVA_CLIENT_SECRET=your_client_secret_here
### 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`).
2. Connect to the server via an MCP Client (like Claude Desktop or MCP Inspector).
3. Call the `get_new_oauth_token` MCP tool.
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!
> **Required OAuth Scopes:**
> `activity:read_all,profile:read_all,read`
1. **Start the server** (it will log a warning that the refresh token is missing, but it will boot!).
2. **Run the Auth Script**:
- Run `uv run auth` in your terminal on your local machine.
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!
---
## Connecting with MCP Clients
The server listens on **port 8000** by default and exposes an SSE endpoint:
`http://localhost:8000/mcp`
The server listens on **port 8000** and exposes an SSE endpoint: `http://localhost:8000/mcp`
### Claude Desktop
Add to your `claude_desktop_config.json`:
Add to `claude_desktop_config.json`:
```json
{
"mcpServers": {
@@ -137,112 +141,88 @@ 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
### Tools
#### 🔐 Authentication
| Tool | Description |
|------|-------------|
| `get_new_oauth_token` | Starts the interactive browser OAuth2 flow to generate and save your initial Refresh Token |
#### 🏃 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).
| Category | Tools |
|----------|-------|
| 🏃 **Athlete** | `get_athlete_profile`, `get_athlete_stats`, `get_athlete_zones` |
| 🚴 **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` |
---
## Project Structure
## Design Decisions
```
strava-mcp-server/
├── 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
```
For a detailed list of architectural choices, unit standardizations, and LLM-specific optimizations, please refer to:
👉 **[docs/DESIGN_DECISIONS.md](docs/DESIGN_DECISIONS.md)**
---
## CI/CD (Gitea Actions)
This repository includes a pre-configured Gitea Action (`.gitea/workflows/cicd.yml`) that automatically:
1. **Lints** the codebase using `ruff` on every push/PR.
2. **Builds & Pushes** the Docker container to the local Container Registry (`git.hnrx.net`) upon a successful merge to `main`.
---
## 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.
Our pipeline (`.gitea/workflows/cicd.yml`) is fully automated:
- **Linting:** Every push/PR is checked with `ruff`.
- **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.
---
## Troubleshooting
### `[Errno 48] Address already in use`
Port 8000 is occupied by a previous server process:
```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`).
`lsof -ti :8000 | xargs kill -9`
### 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 .
```
---
+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.
+2
View File
@@ -40,6 +40,8 @@ Repository = "https://git.hnrx.net/hnrx/strava-mcp-server"
[project.scripts]
strava-mcp = "strava_mcp_server.main: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]
dev = [
+109 -32
View File
@@ -25,33 +25,95 @@ CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET")
REDIRECT_URI = "http://localhost:8765/callback"
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):
client_id: str = ""
client_secret: str = ""
tokens: dict = {}
error: str | None = None
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]
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_header("Content-Type", "text/html")
self.send_header("Content-Type", "text/html; charset=utf-8")
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.</p>
</body></html>
""")
refresh_token = self.tokens.get("refresh_token")
self.wfile.write(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:
error = params.get("error", ["unknown"])[0]
error_msg = 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())
self.wfile.write(f"Error: {error_msg}".encode())
def log_message(self, format, *args):
pass # Suppress server logs
@@ -71,6 +133,12 @@ def main():
f"&scope={SCOPES}"
)
# Configure handler
CallbackHandler.client_id = CLIENT_ID
CallbackHandler.client_secret = CLIENT_SECRET
CallbackHandler.tokens = {}
CallbackHandler.error = None
print("=" * 60)
print(" Strava OAuth2 Authorization")
print("=" * 60)
@@ -85,26 +153,15 @@ def main():
server = HTTPServer(("localhost", 8765), CallbackHandler)
server.handle_request() # Handle exactly one request (the callback)
if not auth_code:
print("No authorization code received.")
if CallbackHandler.error:
print(f"Token exchange failed: {CallbackHandler.error}")
return
print("\nExchanging authorization code for tokens...")
response = httpx.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:
print(f"❌ Token exchange failed: {response.status_code} {response.text}")
if not CallbackHandler.tokens:
print("❌ No tokens received.")
return
data = response.json()
data = CallbackHandler.tokens
refresh_token = data["refresh_token"]
athlete = data.get("athlete", {})
@@ -120,6 +177,26 @@ def main():
print(f"STRAVA_REFRESH_TOKEN={refresh_token}")
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__":
main()
+16 -2
View File
@@ -27,7 +27,7 @@ def validate_credentials() -> None:
if not os.getenv("STRAVA_REFRESH_TOKEN"):
print("️ No STRAVA_REFRESH_TOKEN found. Server starting in unauthenticated mode.")
print(" Use the 'get_new_oauth_token' tool via MCP to authenticate.")
print(" Run 'uv run auth' on your local machine to authenticate.")
def main() -> None:
@@ -42,7 +42,13 @@ def main() -> None:
mcp = FastMCP(
"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,
port=port,
streamable_http_path="/mcp",
@@ -56,12 +62,20 @@ def main() -> None:
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" MCP endpoint: http://{host}:{port}/mcp (Streamable HTTP)")
try:
mcp.run(transport="streamable-http")
except KeyboardInterrupt:
pass
else:
# Run in STDIO mode (default for local testing and Claude Desktop)
mcp.run(transport="stdio")
if __name__ == "__main__":
+1 -1
View File
@@ -37,7 +37,7 @@ class StravaClient:
async def get_valid_token(self) -> str:
"""Returns a valid access token, refreshing it if necessary."""
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:
await self._refresh_access_token()
-2
View File
@@ -14,7 +14,6 @@ from . import segments
from . import segment_efforts
from . import gear
from . import prompts
from . import auth
def register_tools(mcp: FastMCP, strava: StravaClient) -> None:
"""Register all available tools and prompts."""
@@ -26,4 +25,3 @@ def register_tools(mcp: FastMCP, strava: StravaClient) -> None:
segment_efforts.register(mcp, strava)
gear.register(mcp, strava)
prompts.register(mcp, strava)
auth.register(mcp, strava)
+111 -40
View File
@@ -1,7 +1,24 @@
import json
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.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:
@mcp.tool()
@@ -9,16 +26,15 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
ctx: Context,
limit: int = 10,
page: int = 1,
before: int | None = None,
after: int | None = None,
) -> list[TextContent]:
before: str | None = None,
after: str | None = None,
):
"""
List recent Strava activities for the authenticated user.
:param limit: Number of activities to return per page (default 10, max 200).
:param page: Page number for pagination (default 1).
:param before: Unix timestamp — only return activities before this time.
:param after: Unix timestamp — only return activities after this time.
Example: 1704067200 = 2024-01-01 00:00:00 UTC.
:param before: ISO 8601 date string (e.g., '2024-05-01T00:00:00Z') — only return activities before this time.
:param after: ISO 8601 date string (e.g., '2024-04-01T00:00:00Z') — only return activities after this time.
"""
try:
await ctx.info(f"Fetching activities (Page {page}, Limit {limit})...")
@@ -26,27 +42,19 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
activities = await strava.list_activities(
limit=min(limit, 200),
page=page,
before=before,
after=after,
before=parse_iso_to_unix(before),
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 = []
for a in activities:
start_date_raw = a.get("start_date")
essential_data.append({
"id": a["id"],
"name": a["name"],
"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",
"moving_time": f"{a.get('moving_time', 0) / 60:.1f} min",
"total_elevation_gain": f"{a.get('total_elevation_gain', 0):.0f} m",
@@ -62,14 +70,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
markdown_summary += "|-------|-------|------|---------|------|------------|------|\n"
for a in essential_data:
hr = f"{a['average_heartrate']:.0f} bpm" if a['average_heartrate'] else "-"
markdown_summary += f"| {a['start_date']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
markdown_summary += f"| {a['start_date_local']} | {a['sport_type']} | {a['name']} | {a['distance']} | {a['moving_time']} | {a['total_elevation_gain']} | {hr} |\n"
return [
TextContent(type="text", text=markdown_summary.strip()),
TextContent(
type="text",
text=f"Raw Data for Analysis: {json.dumps(essential_data, indent=2)}"
)
_user_text(markdown_summary.strip()),
_resource("internal://activities/list", essential_data),
]
except Exception as e:
error_msg = f"Error listing activities: {str(e)}"
@@ -94,9 +99,41 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
segment["id"] = str(segment["id"])
if "id" in activity:
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:
return f"Error fetching activity details: {str(e)}"
return [TextContent(type="text", text=f"Error fetching activity details: {str(e)}")]
@mcp.tool()
async def get_activity_comments(activity_id: int, limit: int = 30):
@@ -107,17 +144,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
comments = await strava.get_activity_comments(activity_id, per_page=limit)
return [
data = [
{
"id": c.get("id"),
"text": c.get("text"),
"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
]
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:
return f"Error fetching comments: {str(e)}"
return [TextContent(type="text", text=f"Error fetching comments: {str(e)}")]
@mcp.tool()
async def get_activity_kudoers(activity_id: int, limit: int = 30):
@@ -128,7 +172,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
kudoers = await strava.get_activity_kudoers(activity_id, per_page=limit)
return [
data = [
{
"id": k.get("id"),
"name": f"{k.get('firstname')} {k.get('lastname')}",
@@ -137,8 +181,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching kudoers: {str(e)}"
return [TextContent(type="text", text=f"Error fetching kudoers: {str(e)}")]
@mcp.tool()
async def get_laps_by_activity_id(activity_id: int):
@@ -149,7 +202,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
laps = await strava.get_activity_laps(activity_id)
return [
data = [
{
"lap_index": lap.get("lap_index"),
"name": lap.get("name"),
@@ -164,8 +217,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching laps: {str(e)}"
return [TextContent(type="text", text=f"Error fetching laps: {str(e)}")]
@mcp.tool()
async def get_zones_by_activity_id(activity_id: int):
@@ -176,7 +239,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
zones = await strava.get_activity_zones(activity_id)
result = []
data = []
md = "### 💓 Zonen-Verteilung\n\n"
for zone in zones:
buckets = [
{
@@ -187,17 +251,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
for i, b in enumerate(zone.get("distribution_buckets", []))
]
result.append({
zone_data = {
"type": zone.get("type", "unknown"),
"sensor_based": zone.get("sensor_based", False),
"score": zone.get("score"),
"custom_zones": zone.get("custom_zones", False),
"points": zone.get("points"),
"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:
return f"Error fetching activity zones: {str(e)}"
return [TextContent(type="text", text=f"Error fetching activity zones: {str(e)}")]
@mcp.tool()
async def get_activity_streams(
+117 -42
View File
@@ -1,11 +1,12 @@
import json
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.utils import format_date_iso, format_date_human
def register(mcp: FastMCP, strava: StravaClient) -> None:
@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.
Returns name, city, country, follower count, and other profile details.
@@ -16,17 +17,6 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
athlete = await strava.get_athlete()
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}.{d.month}.{d.year}"
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"
@@ -40,8 +30,8 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"measurement_units": athlete.get("measurement_preference"),
"is_premium": athlete.get("premium", False),
"profile_medium": athlete.get("profile_medium"),
"created_at": athlete.get("created_at"),
"updated_at": athlete.get("updated_at"),
"created_at": format_date_iso(athlete.get("created_at")),
"updated_at": format_date_iso(athlete.get("updated_at")),
"bio": athlete.get("bio"),
"follower_count": athlete.get("follower_count"),
"friend_count": athlete.get("friend_count"),
@@ -56,15 +46,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
- Measurement Units: {essential_data['measurement_units'] or 'N/A'}
- Strava Summit Member: {'Yes' if essential_data['is_premium'] else 'No'}
- Profile Image (Medium): {essential_data['profile_medium'] or 'N/A'}
- Joined Strava: {format_date(essential_data['created_at'])}
- Last Updated: {format_date(essential_data['updated_at'])}
- Joined Strava: {format_date_human(essential_data['created_at'])}
- Last Updated: {format_date_human(essential_data['updated_at'])}
""".strip()
return [
TextContent(type="text", text=markdown_summary),
TextContent(
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"])
)
]
@@ -74,24 +73,68 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
return [TextContent(type="text", text=error_msg)]
@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.
Returns zone boundaries for both heart rate and power (if a power meter is configured).
"""
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:
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()
async def get_athlete_stats():
async def get_athlete_stats(ctx: Context):
"""
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.
"""
import json
try:
await ctx.info("Fetching athlete statistics...")
stats = await strava.get_athlete_stats()
def fmt_sport(s: dict) -> dict:
@@ -102,23 +145,55 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"elevation_gain": f"{s.get('elevation_gain', 0):.0f} m",
}
result = {
"all_time": {
"runs": fmt_sport(stats.get("all_run_totals", {})),
"rides": fmt_sport(stats.get("all_ride_totals", {})),
"swims": 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", {})),
},
# Prepare structured data for Markdown
all_time = {
"Laufen": fmt_sport(stats.get("all_run_totals", {})),
"Radfahren": fmt_sport(stats.get("all_ride_totals", {})),
"Schwimmen": fmt_sport(stats.get("all_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:
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}
""")]
+48 -6
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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:
@mcp.tool()
async def list_athlete_clubs():
@@ -10,7 +24,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
clubs = await strava.get_athlete_clubs()
return [
data = [
{
"id": c.get("id"),
"name": c.get("name"),
@@ -22,8 +36,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching clubs: {str(e)}"
return [TextContent(type="text", text=f"Error fetching clubs: {str(e)}")]
@mcp.tool()
async def get_club(club_id: int):
@@ -45,7 +69,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
activities = await strava.get_club_activities(club_id, per_page=limit)
return [
data = [
{
"name": a.get("name"),
"sport_type": a.get("sport_type") or a.get("type"),
@@ -56,8 +80,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching club activities: {str(e)}"
return [TextContent(type="text", text=f"Error fetching club activities: {str(e)}")]
@mcp.tool()
async def get_club_members(club_id: int, limit: int = 30):
@@ -68,7 +101,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
members = await strava.get_club_members(club_id, per_page=limit)
return [
data = [
{
"id": m.get("id"),
"name": f"{m.get('firstname')} {m.get('lastname')}",
@@ -77,5 +110,14 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching club members: {str(e)}"
return [TextContent(type="text", text=f"Error fetching club members: {str(e)}")]
+29 -2
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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:
@mcp.tool()
async def get_gear_by_id(gear_id: str):
@@ -12,7 +26,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
g = await strava.get_gear_by_id(gear_id)
return {
data = {
"id": g.get("id"),
"name": g.get("name"),
"nickname": g.get("nickname"),
@@ -24,5 +38,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"primary": g.get("primary", 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:
return f"Error fetching gear: {str(e)}"
return [TextContent(type="text", text=f"Error fetching gear: {str(e)}")]
+4 -3
View File
@@ -29,11 +29,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
:param weeks: Number of weeks to cover (default 4).
Fetches recent activities and athlete stats to produce a summary report.
"""
import time
after_ts = int(time.time()) - weeks * 7 * 24 * 3600
from datetime import datetime, timedelta, timezone
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 (
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. "
"Also read the get_athlete_stats tool for overall totals.\n\n"
"Structure the report as follows:\n"
+42 -4
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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:
@mcp.tool()
async def get_routes_by_athlete_id(limit: int = 30):
@@ -11,7 +25,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
routes = await strava.get_routes_by_athlete(per_page=min(limit, 200))
return [
data = [
{
"id": str(r.get("id")),
"name": r.get("name"),
@@ -26,8 +40,18 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching routes: {str(e)}"
return [TextContent(type="text", text=f"Error fetching routes: {str(e)}")]
@mcp.tool()
async def get_route_by_id(route_id: str):
@@ -38,7 +62,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
r = await strava.get_route_by_id(route_id)
return {
data = {
"id": str(r.get("id")),
"name": r.get("name"),
"description": r.get("description") or "",
@@ -58,8 +82,22 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
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:
return f"Error fetching route: {str(e)}"
return [TextContent(type="text", text=f"Error fetching route: {str(e)}")]
@mcp.tool()
async def get_route_streams(route_id: str):
+47 -6
View File
@@ -1,5 +1,20 @@
import json
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.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:
@mcp.tool()
@@ -12,12 +27,12 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
e = await strava.get_segment_effort(effort_id)
return {
data = {
"id": str(e.get("id")),
"name": e.get("name"),
"elapsed_time": f"{e.get('elapsed_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",
"average_watts": e.get("average_watts"),
"average_heartrate": e.get("average_heartrate"),
@@ -25,8 +40,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"pr_rank": e.get("pr_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:
return f"Error fetching segment effort: {str(e)}"
return [TextContent(type="text", text=f"Error fetching segment effort: {str(e)}")]
@mcp.tool()
async def list_segment_efforts(
@@ -49,20 +80,30 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
end_date_local=end_date_local,
per_page=limit,
)
return [
data = [
{
"id": str(e.get("id")),
"elapsed_time": f"{e.get('elapsed_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_heartrate": e.get("average_heartrate"),
"pr_rank": e.get("pr_rank"),
}
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:
return f"Error fetching segment efforts: {str(e)}"
return [TextContent(type="text", text=f"Error fetching segment efforts: {str(e)}")]
@mcp.tool()
async def get_segment_effort_streams(
+55 -6
View File
@@ -1,6 +1,20 @@
import json
from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, Annotations, EmbeddedResource, TextResourceContents
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:
@mcp.tool()
async def get_segment(segment_id: int):
@@ -11,7 +25,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
s = await strava.get_segment(segment_id)
return {
data = {
"id": s.get("id"),
"name": s.get("name"),
"activity_type": s.get("activity_type"),
@@ -28,8 +42,25 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"city": s.get("city"),
"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:
return f"Error fetching segment: {str(e)}"
return [TextContent(type="text", text=f"Error fetching segment: {str(e)}")]
@mcp.tool()
async def list_starred_segments(limit: int = 30):
@@ -39,7 +70,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
segments = await strava.get_starred_segments(per_page=limit)
return [
data = [
{
"id": s.get("id"),
"name": s.get("name"),
@@ -51,8 +82,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error fetching starred segments: {str(e)}"
return [TextContent(type="text", text=f"Error fetching starred segments: {str(e)}")]
@mcp.tool()
async def explore_segments(
@@ -68,7 +108,7 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
"""
try:
result = await strava.explore_segments(bounds, activity_type)
return [
data = [
{
"id": s.get("id"),
"name": s.get("name"),
@@ -79,8 +119,17 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
}
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:
return f"Error exploring segments: {str(e)}"
return [TextContent(type="text", text=f"Error exploring segments: {str(e)}")]
@mcp.tool()
async def get_segment_streams(
+60
View File
@@ -0,0 +1,60 @@
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)