MCP_TOGGL_TECHNICAL_SPEC.md•18.9 kB
# MCP Toggl Server - Technical Specification
**Version**: 1.0
**Date**: 2025-10-16
**Status**: Draft
---
## Table of Contents
1. [MCP Tool Specification](#mcp-tool-specification)
2. [Service Architecture](#service-architecture)
3. [API Contracts](#api-contracts)
4. [Error Handling](#error-handling)
5. [Testing Strategy](#testing-strategy)
6. [Deployment & Configuration](#deployment--configuration)
---
## MCP Tool Specification
### Tool: `get_toggl_aggregated_data`
Retrieves aggregated Toggl time tracking data for a given date range and returns it in the standardized toggl_aggregated.json format.
#### Input Schema
```json
{
"type": "object",
"properties": {
"start_date": {
"type": "string",
"format": "date",
"description": "Start date (ISO 8601, inclusive): '2025-10-06'"
},
"end_date": {
"type": "string",
"format": "date",
"description": "End date (ISO 8601, inclusive): '2025-10-13'"
},
"user_emails_filter": {
"type": "array",
"items": {"type": "string"},
"description": "Optional filter to specific users",
"example": ["user1@example.com", "user2@example.com"]
},
"workspace_id": {
"type": "string",
"description": "Optional workspace ID override (defaults to env TOGGL_WORKSPACE_ID)"
}
},
"required": ["start_date", "end_date"]
}
```
#### Output Schema
```json
{
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["success", "error"],
"description": "Operation status"
},
"data": {
"type": "object",
"description": "Toggl aggregated data (see Data Structures)",
"nullable": true
},
"error": {
"type": "object",
"description": "Error details if status is 'error'",
"properties": {
"code": {"type": "string"},
"message": {"type": "string"},
"details": {"type": "object"}
},
"nullable": true
},
"metadata": {
"type": "object",
"properties": {
"processing_time_seconds": {"type": "number"},
"api_calls_made": {"type": "integer"},
"users_fetched": {"type": "integer"},
"entries_parsed": {"type": "integer"}
}
}
}
}
```
#### Example Request
```python
# Via volt-agent calling MCP server
result = mcp_client.call_tool("get_toggl_aggregated_data", {
"start_date": "2025-10-06",
"end_date": "2025-10-13",
"user_emails_filter": ["aleksandr.pylaev@wearevolt.com"]
})
```
#### Example Response
```json
{
"status": "success",
"data": {
"run_id": "mcp-toggl-20251016-143022-abc123",
"aggregated_at": "2025-10-16T14:30:22.123456Z",
"start_date": "2025-10-06",
"end_date": "2025-10-13",
"users": {
"aleksandr.pylaev@wearevolt.com": {
"user_email": "aleksandr.pylaev@wearevolt.com",
"matched_entities": [
{
"entity_database": "Scrum",
"entity_type": "Task",
"entity_id": "456",
"project": "Moneyball",
"duration_seconds": 27594,
"entries_count": 2,
"entries": [
{
"description": "Design user interface",
"duration_seconds": 27594,
"duration_hours": 7.665,
"entry_count": 2
}
]
}
],
"unmatched_activities": [
{
"description": "Team meeting",
"duration_seconds": 4284,
"duration_hours": 1.19,
"entries_count": 1
}
],
"statistics": {
"total_duration_seconds": 31878,
"matched_duration_seconds": 27594,
"unmatched_duration_seconds": 4284,
"total_entries": 3,
"matched_entries": 2,
"unmatched_entries": 1
}
}
},
"statistics": {
"total_users": 1,
"total_matched_entities": 1,
"total_unmatched_activities": 1,
"total_duration_seconds": 31878,
"total_matched_duration_seconds": 27594,
"total_unmatched_duration_seconds": 4284
}
},
"error": null,
"metadata": {
"processing_time_seconds": 45.2,
"api_calls_made": 7,
"users_fetched": 1,
"entries_parsed": 3
}
}
```
---
## Service Architecture
### 3.1 Module Structure
```
mcp-toggl-server/
├── pyproject.toml
├── README.md
├── src/
│ └── toggl_mcp/
│ ├── __init__.py
│ ├── server.py # MCP Server implementation
│ ├── tools/
│ │ └── toggl_tool.py # Tool definition & handler
│ ├── services/
│ │ ├── __init__.py
│ │ ├── toggl_service.py # Toggl API communication
│ │ ├── parser_service.py # Description parsing
│ │ └── aggregator_service.py # Data aggregation
│ ├── models/
│ │ ├── __init__.py
│ │ ├── schemas.py # Pydantic models
│ │ ├── toggl_models.py # Toggl API models
│ │ └── errors.py # Error classes
│ ├── utils/
│ │ ├── __init__.py
│ │ ├── logger.py # Logging configuration
│ │ ├── retry.py # Retry logic & backoff
│ │ └── date_utils.py # Date parsing & validation
│ └── config.py # Configuration loading
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Pytest fixtures
│ ├── unit/
│ │ ├── test_parser_service.py
│ │ ├── test_aggregator_service.py
│ │ └── test_utils.py
│ ├── integration/
│ │ └── test_toggl_integration.py # Mock Toggl API
│ └── fixtures/
│ ├── toggl_responses.json # Mock API responses
│ └── sample_data.json # Sample input/output
└── scripts/
├── run_mcp_server.py
├── test_tool.py # Manual tool testing
└── mock_toggl_server.py # Mock Toggl API for testing
```
### 3.2 Service Responsibilities
#### TogglService (`toggl_service.py`)
- **Purpose**: Communicates with Toggl API
- **Responsibilities**:
- Fetch workspace users
- Fetch time entries (with pagination)
- Handle rate limiting & retries
- Enrich responses with user email data
```python
class TogglService:
def __init__(self, api_token: str, workspace_id: str):
self.api_token = api_token
self.workspace_id = workspace_id
async def get_workspace_users(self) -> list[dict]:
"""Fetch all users in workspace."""
# GET /workspaces/{workspace_id}/users
# Returns: [{id, name, email}, ...]
async def get_time_entries(
self,
start_date: str,
end_date: str,
user_ids: list[str] = None
) -> list[dict]:
"""Fetch time entries for date range, day-by-day."""
# Calls _fetch_day() for each day
# Handles pagination within each day
# Returns: aggregated list of entries
async def _fetch_day(
self,
date: str,
page: int = 1,
user_ids: list[str] = None
) -> tuple[list[dict], bool]:
"""Fetch single day with pagination."""
# GET /workspaces/{workspace_id}/reports
# Returns: (entries, has_more)
async def _fetch_with_backoff(self, ...):
"""Execute request with exponential backoff on 429."""
# Retries: 3 attempts
# Backoff: 60s, 120s, 240s
```
#### ParserService (`parser_service.py`)
- **Purpose**: Parses Toggl descriptions
- **Responsibilities**:
- Extract entity ID pattern: `#(\d+)`
- Extract metadata brackets: `\[([^\]]+)\]`
- Classify matched vs unmatched
- Clean descriptions
```python
class ParserService:
# Compiled regex patterns
ENTITY_PATTERN = re.compile(r'#(\d+)')
METADATA_PATTERN = re.compile(r'\[([^\]]+)\]')
def parse_description(self, description: str) -> ParsedEntry:
"""Parse single description."""
# Returns: ParsedEntry(
# description_clean,
# entity_id,
# entity_database,
# entity_type,
# project,
# is_matched
# )
```
#### AggregatorService (`aggregator_service.py`)
- **Purpose**: Aggregates parsed entries
- **Responsibilities**:
- Group by user
- Group matched entries by (database, type, entity_id)
- Group unmatched by description
- Calculate statistics
```python
class AggregatorService:
def aggregate(
self,
parsed_entries: list[ParsedEntry],
time_entries: list[dict]
) -> AggregatedData:
"""Aggregate entries by user and entity."""
# Returns: AggregatedData structure
# - users: dict[email, UserData]
# - statistics: GlobalStats
```
### 3.3 Data Models (Pydantic)
```python
# models/toggl_models.py
class TimeEntry(BaseModel):
id: int
user_id: int
user_email: str
description: str
start: datetime
stop: datetime
duration: int # seconds
tags: list[str] = []
project_id: int | None = None
project_name: str | None = None
billable: bool = False
class ParsedEntry(BaseModel):
description_clean: str
entity_id: str | None = None
entity_database: str | None = None
entity_type: str | None = None
project: str | None = None
is_matched: bool
class MatchedEntity(BaseModel):
entity_database: str
entity_type: str
entity_id: str
project: str | None = None
duration_seconds: int
entries_count: int
entries: list[EntryGroup]
class EntryGroup(BaseModel):
description: str
duration_seconds: int
duration_hours: float
entry_count: int
class UnmatchedActivity(BaseModel):
description: str
duration_seconds: int
duration_hours: float
entries_count: int
class UserStatistics(BaseModel):
total_duration_seconds: int
matched_duration_seconds: int
unmatched_duration_seconds: int
total_entries: int
matched_entries: int
unmatched_entries: int
class UserData(BaseModel):
user_email: str
matched_entities: list[MatchedEntity]
unmatched_activities: list[UnmatchedActivity]
statistics: UserStatistics
class GlobalStatistics(BaseModel):
total_users: int
total_matched_entities: int
total_unmatched_activities: int
total_duration_seconds: int
total_matched_duration_seconds: int
total_unmatched_duration_seconds: int
class AggregatedData(BaseModel):
run_id: str
aggregated_at: datetime
start_date: str # ISO 8601
end_date: str # ISO 8601
users: dict[str, UserData]
statistics: GlobalStatistics
class ToolResponse(BaseModel):
status: str # "success" or "error"
data: AggregatedData | None = None
error: ErrorDetail | None = None
metadata: ProcessingMetadata
```
---
## API Contracts
### 4.1 Error Handling
#### Error Types
| Error Code | HTTP | Description | Action |
|-----------|------|-------------|--------|
| `INVALID_DATE_FORMAT` | 400 | Start/end date not ISO 8601 | Return error, user fixes |
| `INVALID_DATE_RANGE` | 400 | End date < start date | Return error, user fixes |
| `AUTH_FAILED` | 401 | Invalid Toggl API token | Check configuration |
| `WORKSPACE_NOT_FOUND` | 404 | Workspace ID doesn't exist | Check configuration |
| `RATE_LIMIT_EXCEEDED` | 429 | Too many API calls | Retry with backoff |
| `API_ERROR` | 500+ | Toggl API error | Retry with backoff |
| `UNKNOWN` | 500 | Unexpected error | Log & investigate |
#### Error Response Format
```json
{
"status": "error",
"data": null,
"error": {
"code": "INVALID_DATE_FORMAT",
"message": "Start date must be ISO 8601 format (YYYY-MM-DD)",
"details": {
"received": "10-06-2025",
"expected_format": "YYYY-MM-DD"
}
},
"metadata": {
"processing_time_seconds": 0.1
}
}
```
### 4.2 Validation Rules
**Input Validation**:
- `start_date` must be ISO 8601 (YYYY-MM-DD)
- `end_date` must be ISO 8601 (YYYY-MM-DD)
- `end_date >= start_date`
- Date range max: 90 days (to prevent excessive API calls)
- `user_emails_filter`: valid email format if provided
**Processing**:
- Date range internally converted to UTC
- Entries with duration <= 0 filtered out (safety)
- Parse errors logged but non-blocking
---
## Error Handling
### 5.1 Retry Strategy
```python
# RetryConfig
class RetryConfig:
max_attempts: int = 3
initial_backoff_seconds: int = 60
max_backoff_seconds: int = 300
exponential_base: float = 2.0
# Backoff calculation: min(initial_backoff * (base ** attempt), max_backoff)
# Attempt 1: 60s
# Attempt 2: 120s
# Attempt 3: 240s
```
### 5.2 Graceful Degradation
- **Partial User Failures**: If one user fails, continue with others
- **Parse Errors**: Log as warning, continue (non-blocking)
- **Missing Fields**: Use defaults (e.g., project=None)
### 5.3 Logging
```python
# Logging levels
logger.debug(f"Fetching entries for user {user_id}")
logger.info(f"Parsed {total_entries} entries")
logger.warning(f"Rate limited, retrying after {wait_time}s")
logger.error(f"Failed to fetch user {user_id}: {error}")
```
---
## Testing Strategy
### 6.1 Unit Tests
#### ParserService Tests
```python
def test_parse_matched_entry_with_all_fields():
parser = ParserService()
result = parser.parse_description("Design UI #456 [Scrum] [Task] [Moneyball]")
assert result.description_clean == "Design UI"
assert result.entity_id == "456"
assert result.entity_database == "Scrum"
assert result.entity_type == "Task"
assert result.project == "Moneyball"
assert result.is_matched is True
def test_parse_unmatched_entry():
parser = ParserService()
result = parser.parse_description("Team meeting")
assert result.description_clean == "Team meeting"
assert result.entity_id is None
assert result.is_matched is False
def test_parse_rightmost_entity_id():
parser = ParserService()
result = parser.parse_description("Work #123 then #456 [Scrum]")
assert result.entity_id == "456" # Rightmost wins
def test_parse_whitespace_and_newlines():
parser = ParserService()
result = parser.parse_description("Design\nUI \n#456\n[Scrum]")
assert result.entity_id == "456"
assert result.entity_database == "Scrum"
```
#### AggregatorService Tests
```python
def test_aggregate_by_user():
# Input: parsed entries for 2 users
# Expected: users dict with correct grouping
def test_aggregate_matched_by_entity():
# Input: multiple entries for same entity
# Expected: entries grouped by (db, type, id)
def test_aggregate_unmatched_by_description():
# Input: multiple unmatched entries with same description
# Expected: entries grouped by description
def test_statistics_calculations():
# Verify totals and counts are correct
```
### 6.2 Integration Tests
Mock the Toggl API:
```python
@pytest.fixture
def mock_toggl_api(monkeypatch):
"""Mock Toggl API responses."""
def mock_response(*args, **kwargs):
return {
"data": [
{
"user_id": 1,
"user_email": "test@example.com",
"description": "Task #123 [Scrum]",
"duration": 3600
}
],
"has_more": False
}
monkeypatch.setattr(
"toggl_mcp.services.toggl_service.TogglClient.get_time_entries",
mock_response
)
def test_full_pipeline(mock_toggl_api):
"""Test complete pipeline from input to output."""
service = TogglService(api_token="test", workspace_id="123")
result = service.aggregate("2025-10-06", "2025-10-13")
assert result.status == "success"
assert len(result.data.users) > 0
assert result.data.statistics.total_users > 0
```
### 6.3 Performance Tests
```python
def test_performance_large_dataset():
"""Ensure processing completes within time budget."""
# Input: 10,000 entries across 100 users
# Expected: Processing completes in < 5 seconds
start = time.time()
result = aggregator.aggregate(large_entry_set)
elapsed = time.time() - start
assert elapsed < 5.0
```
---
## Deployment & Configuration
### 7.1 MCP Server Startup
```python
# run_mcp_server.py
from toggl_mcp.server import TogglMCPServer
async def main():
server = TogglMCPServer()
await server.start()
if __name__ == "__main__":
asyncio.run(main())
```
### 7.2 Environment Configuration
```bash
# Required
export TOGGL_API_TOKEN="your_token_here"
export TOGGL_WORKSPACE_ID="1637944"
# Optional
export MCP_LOG_LEVEL="info"
export MCP_PORT="8001"
export TOGGL_API_BASE_URL="https://api.track.toggl.com/reports/api/v3"
export TOGGL_RETRY_MAX_ATTEMPTS="3"
export TOGGL_RETRY_INITIAL_BACKOFF="60"
```
### 7.3 Docker Deployment
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY pyproject.toml .
RUN pip install -e .
COPY src ./src
ENV TOGGL_API_TOKEN=""
ENV TOGGL_WORKSPACE_ID=""
ENV MCP_LOG_LEVEL="info"
CMD ["python", "-m", "toggl_mcp.server"]
```
### 7.4 Integration with volt-agent
```python
# In volt-agent workflow
from mcp_client import MCPClient
client = MCPClient(
server_path="path/to/mcp_toggl_server",
transport="stdio" # or "http"
)
result = client.call_tool("get_toggl_aggregated_data", {
"start_date": "2025-10-06",
"end_date": "2025-10-13"
})
# result = {
# "status": "success",
# "data": {...},
# "metadata": {...}
# }
```
---
## Appendix: Development Checklist
### Code Quality
- [ ] Type hints on all functions
- [ ] Docstrings on all modules/classes/functions
- [ ] Error handling for all edge cases
- [ ] Logging at appropriate levels
- [ ] No hardcoded secrets
### Testing
- [ ] Unit tests for each service (>80% coverage)
- [ ] Integration tests with mock Toggl API
- [ ] Performance tests on large datasets
- [ ] Error scenario tests
### Documentation
- [ ] README with setup instructions
- [ ] API documentation (this file)
- [ ] Code comments on complex logic
- [ ] Example usage scripts
### Security
- [ ] API token from environment (not hardcoded)
- [ ] Rate limiting respected
- [ ] Input validation on all parameters
- [ ] Error messages don't leak sensitive info
### Deployment
- [ ] Docker image builds successfully
- [ ] Environment variables documented
- [ ] Startup verification script
- [ ] Health check endpoint
---
## References
- [Toggl Reports API v3](https://engineering.toggl.com/docs/reports/)
- [MCP Specification](https://spec.modelcontextprotocol.io/)
- [Pydantic Documentation](https://docs.pydantic.dev/)
- [AsyncIO Guide](https://docs.python.org/3/library/asyncio.html)