MCP Cosmotech API Client

A Python client library for interacting with the Cosmotech API using the Model Context Protocol (MCP).

Features

• OAuth2 authentication management
• Organization management tools
• Workspace management tools
• Solution management tools with parameter and run template support
• Runner management tools with resource and configuration management
• Async/await support
• Automatic pagination handling
• Comprehensive error handling

Requirements

• Python 3.9 or higher
• Poetry for dependency management

Installation

Using Poetry (recommended)

poetry add mcp-cosmotech

Using pip

pip install mcp-cosmotech

Quick Start

Environment Setup

Set the following environment variables:

export COSMOTECH_API_CLIENT_ID="your_client_id"
export COSMOTECH_API_CLIENT_SECRET="your_client_secret"
export COSMOTECH_API_TOKEN_URL="https://auth.cosmotech.com/token"
export COSMOTECH_API_BASE_URL="https://api.cosmotech.com"

Full Workflow Example

For a comprehensive example demonstrating the complete workflow including:

• Organization creation and management
• Workspace setup within organizations
• Solution registration and configuration
• Runner setup and management

Check out our Full Workflow Example.

Basic Usage Example

Here's a simple example of using the organization, workspace, and solution management tools:

import asyncio
from mcp_cosmotech.utils.auth import OAuth2Handler
from mcp_cosmotech.utils.client import APIClient
from mcp_cosmotech.tools.organizations import OrganizationTools, OrganizationCreate
from mcp_cosmotech.tools.workspaces import WorkspaceTools, WorkspaceCreate
from mcp_cosmotech.tools.solutions import (
    SolutionTools,
    SolutionCreate,
    SolutionParameter,
    ParameterGroup,
    RunTemplate
)
from mcp_cosmotech.tools.runners import (
    RunnerTools,
    RunnerCreate
)

async def main():
    # Setup authentication
    auth_handler = OAuth2Handler(
        client_id="your_client_id",
        client_secret="your_client_secret",
        token_url="https://auth.cosmotech.com/token"
    )
    
    # Initialize client and tools
    client = APIClient(auth_handler=auth_handler)
    org_tools = OrganizationTools(client)
    workspace_tools = WorkspaceTools(client)
    solution_tools = SolutionTools(client)
    runner_tools = RunnerTools(client)
    
    # Create organization
    org = await org_tools.create_organization(
        OrganizationCreate(
            name="My Organization",
            owner_email="admin@example.com",
            description="My first organization"
        )
    )
    print(f"Created organization: {org.name} (ID: {org.id})")

    # Create workspace in the organization
    workspace = await workspace_tools.create_workspace(
        org.id,
        WorkspaceCreate(
            name="My Workspace",
            description="My first workspace",
            key="my-workspace",
            tags=["example"]
        )
    )
    print(f"Created workspace: {workspace.name} (ID: {workspace.id})")

    # Create and register a solution
    solution = await solution_tools.register_solution(
        org.id,
        SolutionCreate(
            name="My Solution",
            description="Example solution",
            version="1.0.0",
            parameters={
                "param1": SolutionParameter(
                    id="param1",
                    name="Parameter 1",
                    type="string",
                    description="Example parameter"
                )
            }
        )
    )
    print(f"Registered solution: {solution.name} (ID: {solution.id})")
    
    # Create a runner in the organization
    runner = await runner_tools.create_runner(
        org.id,
        RunnerCreate(
            name="My Runner",
            description="Example runner",
            resource_requirements={
                "cpu": "2",
                "memory": "4Gi"
            },
            configuration={
                "env": {"LOG_LEVEL": "INFO"}
            }
        )
    )
    print(f"Created runner: {runner.name} (ID: {runner.id}, Status: {runner.status})")

if __name__ == "__main__":
    asyncio.run(main())

Documentation

• Organization Management Tools - Complete guide to managing organizations
• Workspace Management Tools - Complete guide to managing workspaces
• Solution Management Tools - Complete guide to managing solutions, parameters, and run templates
• Runner Management Tools - Complete guide to managing runners, resources, and configurations
• Examples - Example scripts demonstrating API usage including:
  • Basic organization management (organizations.py)
  • Workspace management (workspaces.py)
  • Solution management (solutions.py)
  • Runner management (runners.py)
  • Complete workflow integration (full_workflow.py)

Error Handling

The library provides two main exception types:

• AuthenticationError - For authentication-related errors
• APIError - For API request errors

Example error handling:

from mcp_cosmotech.utils.client import APIError, AuthenticationError

try:
    solution = await solution_tools.get_solution("org_id", "solution_id")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except APIError as e:
    print(f"API error: {e}, Status: {e.status_code}")

Dependencies

Core dependencies:

• pydantic (^2.0) - Data validation and settings management
• aiohttp (^3.8) - Async HTTP client
• typing-extensions (^4.0) - Enhanced typing support

Development dependencies:

• pytest and pytest-asyncio - Testing framework
• black - Code formatting
• mypy - Static type checking
• ruff - Python linter

Development Setup

1. Clone the repository:

git clone https://github.com/Cosmo-Tech/mcp-cosmotech-api.git
cd mcp-cosmotech-api

2. Install dependencies with Poetry:

poetry install

3. Run tests:

poetry run pytest

License

[License information to be added]

Contributing

[Contributing guidelines to be added]