Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
on:
release:
types: [ "published" ]
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]

jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3

- name: Install just
run: sudo apt-get update && sudo apt-get install -y just

- name: Install uv
uses: astral-sh/setup-uv@7edac99f961f18b581bbd960d59d049f04c0002f # v6.4.1
with:
enable-cache: true
cache-dependency-glob: "uv.lock"

- name: Install python packages
run: uv sync --locked --all-extras --dev

- name: Test
run: just test

- name: Lint
run: just lint

- name: Typecheck
run: just typecheck
18 changes: 18 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,24 @@ Aptible MCP is a Mission Control Protocol (MCP) integration for interacting with
python main.py
```

### Development Commands

```bash
# Run tests
just test

# Run a single test file
APTIBLE_TOKEN="foobar" APTIBLE_API_ROOT_URL="http://localhost:3000" APTIBLE_AUTH_ROOT_URL="http://localhost:3001" python -m pytest -v -s tests/test_file.py

# Run linting
just lint

# Format code and sort dependencies
just pretty

# Run type checking
just typecheck
```

## Authentication

Expand Down
84 changes: 45 additions & 39 deletions api_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,51 +8,52 @@
from typing import Any, Dict, Optional


APTIBLE_API_ROOT_URL = os.environ.get("APTIBLE_API_ROOT_URL", "https://api.aptible.com")
APTIBLE_AUTH_ROOT_URL = os.environ.get("APTIBLE_AUTH_ROOT_URL", "https://auth.aptible.com")
APTIBLE_TOKEN = os.environ.get("APTIBLE_TOKEN", None)


class AptibleApiClient:
"""
Aptible API client for making authenticated requests to the API.
"""
def __init__(self, api_url: Optional[str] = None, auth_url: Optional[str] = None) -> None:
self.api_url = api_url or APTIBLE_API_ROOT_URL
self.auth_url = auth_url or APTIBLE_AUTH_ROOT_URL
self._token = None


def __init__(
self, api_url: Optional[str] = None, auth_url: Optional[str] = None
) -> None:
self.api_url = api_url or os.environ.get(
"APTIBLE_API_ROOT_URL", "https://api.aptible.com"
)
self.auth_url = auth_url or os.environ.get(
"APTIBLE_AUTH_ROOT_URL", "https://auth.aptible.com"
)
self._token: Optional[str] = os.environ.get("APTIBLE_TOKEN", None)

def get_token(self) -> str:
"""
Get authentication token.
"""
if self._token:
return self._token

# Try to get from environment
if APTIBLE_TOKEN:
self._token = APTIBLE_TOKEN
return self._token

# Try to get from file

# Try to get from ~/.aptible/tokens.json file
home = Path.home()
try:
with open(home / ".aptible" / "tokens.json") as f:
data = json.load(f)
self._token = data[self.auth_url]
except (FileNotFoundError, KeyError) as e:
raise Exception("Authentication token not found. Please login to Aptible CLI first.")

except (FileNotFoundError, KeyError):
raise Exception(
"Authentication token not found. Please login to Aptible CLI first."
)

if not self._token:
raise Exception("You are not logged in")

return self._token

def fetch_public_key(self) -> str:
"""
Gets the public key used for signing JWTs
from the Aptible Auth API.
"""
if not self.auth_url:
raise ValueError("Auth URL is not set")
response = requests.get(self.auth_url)
public_key = response.json()["public_key"]
return public_key
Expand All @@ -63,15 +64,16 @@ def parsed_token(self) -> Dict[str, Any]:
"""
token = self.get_token()
public_key = self.fetch_public_key()
kwargs = {
'algorithms': ["RS256", "RS512"],
'options': {
'verify_signature': True,
'verify_exp': True,
return jwt.decode(
token,
public_key,
algorithms=["RS256", "RS512"],
options={
"verify_signature": True,
"verify_exp": True,
},
'leeway': 0,
}
return jwt.decode(token, public_key, **kwargs)
leeway=0,
)

def organization_id(self) -> str:
"""
Expand All @@ -83,7 +85,9 @@ def organization_id(self) -> str:
it's just here right now because that's where the token is.
If a better place for this arises, please move it!
"""
response = requests.get(f"{self.auth_url}/organizations", headers=self._get_headers())
response = requests.get(
f"{self.auth_url}/organizations", headers=self._get_headers()
)
response.raise_for_status()
orgs = response.json()["_embedded"]["organizations"]
if orgs == 0:
Expand All @@ -98,7 +102,7 @@ def _get_headers(self) -> Dict[str, str]:
"Content-Type": "application/hal+json",
"Authorization": f"Bearer {self.get_token()}",
}

def _build_url(self, path: str) -> str:
"""
Build the full URL for the API from just a path.
Expand All @@ -113,7 +117,7 @@ def get(self, path: str) -> Any:
response = requests.get(url, headers=self._get_headers())
response.raise_for_status()
return response.json()

def post(self, path: str, data: Any) -> Any:
"""
Make a POST request to Aptible API.
Expand All @@ -122,7 +126,7 @@ def post(self, path: str, data: Any) -> Any:
response = requests.post(url, headers=self._get_headers(), json=data)
response.raise_for_status()
return response.json()

def put(self, path: str, data: Any) -> Any:
"""
Make a PUT request to Aptible API.
Expand All @@ -131,21 +135,21 @@ def put(self, path: str, data: Any) -> Any:
response = requests.put(url, headers=self._get_headers(), json=data)
response.raise_for_status()
return response.json()

def delete(self, path: str) -> Any:
"""
Make a DELETE request to Aptible API.
"""
url = self._build_url(path)
response = requests.delete(url, headers=self._get_headers())
response.raise_for_status()

# Some DELETE responses may not return content
if response.status_code == 204 or not response.content:
return None

return response.json()

def wait_for_operation(self, operation_id: str) -> None:
"""
Waits on the operation to reach a completed state
Expand All @@ -164,5 +168,7 @@ def wait_for_operation(self, operation_id: str) -> None:
status = response.get("status", "unknown")
if status in done_states:
if status == "failed":
raise Exception(f"Operation {operation_id} failed: {response.get('message', 'No error message')}")
return None
raise Exception(
f"Operation {operation_id} failed: {response.get('message', 'No error message')}"
)
return None
18 changes: 18 additions & 0 deletions justfile
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
lint:
uv run ruff check
uv run ruff format --check
uv run toml-sort --check pyproject.toml

pretty:
uv run ruff check --fix
uv run ruff format
uv run toml-sort pyproject.toml

test:
APTIBLE_TOKEN="foobar" \
APTIBLE_API_ROOT_URL="http://localhost:3000" \
APTIBLE_AUTH_ROOT_URL="http://localhost:3001" \
uv run python -m pytest -v -s tests/

typecheck:
uv run mypy --show-traceback .
Loading