from pydantic import BaseModel, Field, validator
from typing import Dict, Any, Optional, Union
from datetime import datetime
import uuid
class TokenData(BaseModel):
"""
Pydantic model for JWT token data
"""
user_id: int = Field(..., description="Authenticated user ID from JWT token")
class Config:
schema_extra = {
"example": {
"user_id": 12345
}
}
class ExecuteRequest(BaseModel):
"""
Pydantic model for MCP execute endpoint request
"""
service_name: str = Field(
...,
description="Name of the service to execute (github, vercel, linear, etc.)",
min_length=1,
max_length=50
)
payload: Dict[str, Any] = Field(
...,
description="Request payload data for the service operation",
min_items=1
)
@validator('service_name')
def validate_service_name(cls, v):
"""Validate service name"""
valid_services = {
'github', 'vercel', 'linear', 'notion', 'slack',
'discord', 'google_workspace'
}
if v.lower() not in valid_services:
raise ValueError(
f"Invalid service name. Must be one of: {', '.join(sorted(valid_services))}"
)
return v.lower()
class Config:
schema_extra = {
"example": {
"service_name": "github",
"payload": {
"action": "create_issue",
"repository": "my-org/my-repo",
"title": "Bug fix needed",
"body": "There's a bug in the login functionality",
"labels": ["bug", "high-priority"]
}
}
}
class MCPResponse(BaseModel):
"""
Pydantic model for MCP response
"""
status: str = Field(..., description="Response status (success, error, pending)")
action: str = Field(..., description="Action that was performed")
data: Dict[str, Any] = Field(..., description="Response data from the service")
message: str = Field(..., description="Human-readable message")
service: str = Field(..., description="Service name that handled the request")
user_id: int = Field(..., description="User ID who made the request")
timestamp: str = Field(..., description="ISO timestamp of the response")
class Config:
schema_extra = {
"example": {
"status": "success",
"action": "github_issue_created",
"data": {
"issue_id": "123",
"url": "https://github.com/my-org/my-repo/issues/123",
"number": 42
},
"message": "GitHub issue created successfully",
"service": "github",
"user_id": 12345,
"timestamp": "2025-11-01T14:15:22Z"
}
}
class ErrorResponse(BaseModel):
"""
Pydantic model for error responses
"""
error: str = Field(..., description="Error type or code")
detail: str = Field(..., description="Detailed error message")
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
class Config:
schema_extra = {
"example": {
"error": "VALIDATION_ERROR",
"detail": "Invalid service name provided",
"timestamp": "2025-11-01T14:15:22Z"
}
}
class HealthResponse(BaseModel):
"""
Pydantic model for health check response
"""
status: str = Field(..., description="Health status")
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
version: str = Field(default="1.0.0", description="Application version")
class Config:
schema_extra = {
"example": {
"status": "ok",
"timestamp": "2025-11-01T14:15:22Z",
"version": "1.0.0"
}
}
class ServicesListResponse(BaseModel):
"""
Pydantic model for services list response
"""
services: list = Field(..., description="List of available services")
count: int = Field(..., description="Total number of services")
class Config:
schema_extra = {
"example": {
"services": ["github", "vercel", "linear", "notion", "slack", "discord", "google_workspace"],
"count": 7
}
}
class MCPStatusResponse(BaseModel):
"""
Pydantic model for MCP status response
"""
status: str = Field(..., description="Service status")
version: str = Field(..., description="MCP version")
services_count: int = Field(..., description="Number of supported services")
uptime: str = Field(..., description="Uptime information")
class Config:
schema_extra = {
"example": {
"status": "active",
"version": "1.0.0",
"services_count": 7,
"uptime": "active"
}
}
class UserCredentials(BaseModel):
"""
Pydantic model for user credentials
"""
service_name: str = Field(..., description="Name of the service")
credentials: Dict[str, Any] = Field(..., description="Encrypted credential data")
created_at: Optional[datetime] = Field(None, description="When credentials were created")
updated_at: Optional[datetime] = Field(None, description="When credentials were last updated")
@validator('service_name')
def validate_service_name(cls, v):
"""Validate service name"""
valid_services = {
'github', 'vercel', 'linear', 'notion', 'slack',
'discord', 'google_workspace'
}
if v.lower() not in valid_services:
raise ValueError(
f"Invalid service name. Must be one of: {', '.join(sorted(valid_services))}"
)
return v.lower()
class Config:
schema_extra = {
"example": {
"service_name": "github",
"credentials": {
"encrypted_token": "gAAAAABh...",
"username": "user123"
},
"created_at": "2025-11-01T14:15:22Z",
"updated_at": "2025-11-01T14:15:22Z"
}
}
class CredentialsValidationRequest(BaseModel):
"""
Pydantic model for credentials validation request
"""
service_name: str = Field(..., description="Name of the service to validate")
credentials: Dict[str, Any] = Field(..., description="Credentials to validate")
@validator('service_name')
def validate_service_name(cls, v):
"""Validate service name"""
valid_services = {
'github', 'vercel', 'linear', 'notion', 'slack',
'discord', 'google_workspace'
}
if v.lower() not in valid_services:
raise ValueError(
f"Invalid service name. Must be one of: {', '.join(sorted(valid_services))}"
)
return v.lower()
class Config:
schema_extra = {
"example": {
"service_name": "github",
"credentials": {
"token": "ghp_xxxxxxxxxxxx",
"username": "myusername"
}
}
}
class CredentialsValidationResponse(BaseModel):
"""
Pydantic model for credentials validation response
"""
valid: bool = Field(..., description="Whether credentials are valid")
service_name: str = Field(..., description="Validated service name")
message: str = Field(..., description="Validation message")
metadata: Optional[Dict[str, Any]] = Field(None, description="Additional validation metadata")
class Config:
schema_extra = {
"example": {
"valid": True,
"service_name": "github",
"message": "Credentials validated successfully",
"metadata": {
"username": "myusername",
"scope": "repo,user"
}
}
}
