Compare commits
7 Commits
v0.0.4
...
bcc11cb07e
| Author | SHA1 | Date | |
|---|---|---|---|
| bcc11cb07e | |||
| 40b5d004b1 | |||
| 7c089d90c5 | |||
| c69e362635 | |||
| 3805ca3274 | |||
| c56f7ad7b4 | |||
| 578c4b292a |
+5
-3
@@ -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"]
|
||||
|
||||
@@ -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 .
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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.
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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]
|
||||
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>✅ Authorization successful!</h2>
|
||||
<p>You can close this window and return to your terminal.</p>
|
||||
</body></html>
|
||||
""")
|
||||
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; charset=utf-8")
|
||||
self.end_headers()
|
||||
|
||||
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">✅</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;">
|
||||
— Strava MCP Server Authorization Helper —
|
||||
</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>❌ 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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
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
|
||||
# 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__":
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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>✅ 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>❌ 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}
|
||||
""")]
|
||||
@@ -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)}")]
|
||||
|
||||
@@ -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)}")]
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user