File size: 3,841 Bytes
5da4770 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | from typing import Protocol, List, Optional, Dict, Any
from uuid import UUID
from .domain.entities import Profile, Connection, App, MCPServer
from .domain.value_objects import ExternalUserId, AppSlug, ProfileName, Category, SearchQuery, PaginationCursor
class Logger(Protocol):
def info(self, message: str) -> None: ...
def warning(self, message: str) -> None: ...
def error(self, message: str) -> None: ...
class DatabaseConnection(Protocol):
async def client(self) -> Any: ...
class ProfileRepository(Protocol):
async def create(self, profile: Profile) -> Profile: ...
async def get_by_id(self, account_id: UUID, profile_id: UUID) -> Optional[Profile]: ...
async def get_by_app_slug(self, account_id: UUID, app_slug: AppSlug, profile_name: Optional[ProfileName] = None) -> Optional[Profile]: ...
async def find_by_account(self, account_id: UUID, app_slug: Optional[AppSlug] = None, is_active: Optional[bool] = None) -> List[Profile]: ...
async def update(self, profile: Profile) -> Profile: ...
async def delete(self, account_id: UUID, profile_id: UUID) -> bool: ...
async def set_default(self, account_id: UUID, profile_id: UUID, mcp_qualified_name: str) -> None: ...
class ConnectionRepository(Protocol):
async def get_by_external_user_id(self, external_user_id: ExternalUserId) -> List[Connection]: ...
async def create(self, connection: Connection) -> Connection: ...
async def update(self, connection: Connection) -> Connection: ...
async def delete(self, external_user_id: ExternalUserId, app_slug: AppSlug) -> bool: ...
class AppRepository(Protocol):
async def search(self, query: SearchQuery, category: Optional[Category] = None,
page: int = 1, limit: int = 20, cursor: Optional[PaginationCursor] = None) -> Dict[str, Any]: ...
async def get_by_slug(self, app_slug: AppSlug) -> Optional[App]: ...
async def get_icon_url(self, app_slug: AppSlug) -> Optional[str]: ...
async def get_popular(self, category: Optional[Category] = None, limit: int = 10) -> List[App]: ...
async def get_by_category(self, category: Category, limit: int = 20) -> List[App]: ...
class MCPServerRepository(Protocol):
async def discover_for_user(self, external_user_id: ExternalUserId, app_slug: Optional[AppSlug] = None) -> List[MCPServer]: ...
async def test_connection(self, server: MCPServer) -> MCPServer: ...
async def create_connection(self, external_user_id: ExternalUserId, app_slug: AppSlug, oauth_app_id: Optional[str] = None) -> MCPServer: ...
class ExternalUserIdGeneratorService(Protocol):
def generate(self, account_id: str, app_slug: AppSlug, profile_name: ProfileName) -> ExternalUserId: ...
class MCPQualifiedNameService(Protocol):
def generate(self, app_slug: AppSlug) -> str: ...
class ConnectionTokenService(Protocol):
async def create(self, external_user_id: ExternalUserId, app: Optional[AppSlug] = None) -> Dict[str, Any]: ...
class ProfileConfigurationService(Protocol):
def validate_config(self, config: Dict[str, Any]) -> bool: ...
def merge_config(self, existing_config: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: ...
class ConnectionStatusService(Protocol):
async def check_connection_status(self, profile: Profile) -> bool: ...
async def update_connection_status(self, profile: Profile) -> Profile: ...
class HttpClient(Protocol):
async def get(self, url: str, headers: Dict[str, str] = None, params: Dict[str, Any] = None) -> Dict[str, Any]: ...
async def post(self, url: str, headers: Dict[str, str] = None, json: Dict[str, Any] = None) -> Dict[str, Any]: ...
async def close(self) -> None: ...
class EncryptionService(Protocol):
def encrypt(self, data: str) -> str: ...
def decrypt(self, encrypted_data: str) -> str: ... |