7 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
18 changed files with 776 additions and 385 deletions
+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"]
+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) - 🛠️ **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,9 +27,10 @@ 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)
- [Known Strava API Limitations](#known-strava-api-limitations) - [Known Strava API Limitations](#known-strava-api-limitations)
- [Troubleshooting](#troubleshooting) - [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 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 +49,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 +65,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 +114,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 +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 ## 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 .
```
--- ---
+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] [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 = [
+109 -32
View File
@@ -25,33 +25,95 @@ 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(f"""
</body></html> <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 +133,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 +153,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 +177,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()
+16 -2
View File
@@ -27,7 +27,7 @@ def validate_credentials() -> None:
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("️ 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: def main() -> None:
@@ -42,7 +42,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 +62,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__":
+1 -1
View File
@@ -37,7 +37,7 @@ 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()
-2
View File
@@ -14,7 +14,6 @@ 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 +25,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)
+111 -40
View File
@@ -1,7 +1,24 @@
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 +26,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,27 +42,19 @@ 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:
start_date_raw = a.get("start_date")
essential_data.append({ 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",
@@ -62,14 +70,11 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
markdown_summary += "|-------|-------|------|---------|------|------------|------|\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 = 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 [ 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 +99,41 @@ 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 +144,24 @@ 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 +172,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 +181,17 @@ 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 +202,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 +217,18 @@ 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 +239,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 +251,24 @@ 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(
+117 -42
View File
@@ -1,11 +1,12 @@
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.
@@ -16,17 +17,6 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
athlete = await strava.get_athlete() 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_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"
@@ -40,8 +30,8 @@ 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"),
@@ -56,15 +46,24 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
- 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"])
) )
] ]
@@ -74,24 +73,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 +145,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}
""")]
+48 -6
View File
@@ -1,6 +1,20 @@
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 +24,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 +36,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 +69,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 +80,17 @@ 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 +101,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 +110,14 @@ 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)}")]
+29 -2
View File
@@ -1,6 +1,20 @@
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 +26,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 +38,18 @@ 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)}")]
+4 -3
View File
@@ -29,11 +29,12 @@ 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"
+42 -4
View File
@@ -1,6 +1,20 @@
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,7 +25,7 @@ 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"),
@@ -26,8 +40,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,7 +62,7 @@ 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 "",
@@ -58,8 +82,22 @@ 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):
+47 -6
View File
@@ -1,5 +1,20 @@
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 +27,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 +40,24 @@ 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 +80,30 @@ 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(
+55 -6
View File
@@ -1,6 +1,20 @@
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 +25,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 +42,25 @@ 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 +70,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 +82,17 @@ 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 +108,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 +119,17 @@ 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(
+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)