Skip to content
Closed
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
17 changes: 17 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,20 @@ _bmad-output/*
._*
*.bak
server
<<<<<<< HEAD
=======
server
cli-proxy-api-plus-integration-test

boardsync
releasebatch
.cache
>>>>>>> a4e4c2b8 (chore: add build artifacts to .gitignore)

# Build artifacts (cherry-picked from fix/test-cleanups)
cliproxyapi++
.air/
boardsync
releasebatch
.cache
logs/
Comment on lines +57 to +73
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

This change introduces merge conflict markers which should be resolved. It also adds duplicate entries. Please resolve the conflicts and remove duplicates.


# Build artifacts (cherry-picked from fix/test-cleanups)
cli-proxy-api-plus-integration-test
cliproxyapi++
.air/
boardsync
releasebatch
.cache
logs/

276 changes: 276 additions & 0 deletions sdk/python/cliproxy/api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,276 @@
"""
Comprehensive Python SDK for cliproxyapi-plusplus.

NOT just HTTP wrappers - provides native Python classes and functions.
Translates Go types to Python dataclasses with full functionality.
"""

import httpx
from dataclasses import dataclass, field
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To use asdict for dataclass serialization, it needs to be imported from the dataclasses module.

Suggested change
from dataclasses import dataclass, field
from dataclasses import dataclass, field, asdict

from typing import Any, Optional
from enum import Enum
import os
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To avoid hardcoding the default model name in multiple places, it's good practice to define it as a constant.

Suggested change
import os
import os
DEFAULT_MODEL = "claude-3-5-sonnet-20241022"



# =============================================================================
# Enums - Native Python
# =============================================================================

class ModelProvider(str, Enum):
"""Supported model providers."""
OPENAI = "openai"
ANTHROPIC = "anthropic"
GOOGLE = "google"
OPENROUTER = "openrouter"
MINIMAX = "minimax"
KIRO = "kiro"
CODEX = "codex"
CLAUDE = "claude"
GEMINI = "gemini"
VERTEX = "vertex"


# =============================================================================
# Models - Native Python classes
# =============================================================================

@dataclass
class ProviderConfig:
"""Native Python config for providers."""
provider: ModelProvider
api_key: Optional[str] = None
base_url: Optional[str] = None
models: list[str] = field(default_factory=list)
timeout: int = 30
max_retries: int = 3


@dataclass
class AuthEntry:
"""Authentication entry."""
name: str
provider: ModelProvider
credentials: dict[str, Any] = field(default_factory=dict)
enabled: bool = True


@dataclass
class ChatMessage:
"""Chat message with role support."""
role: str # "system", "user", "assistant"
content: str
name: Optional[str] = None
Comment on lines +57 to +62
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The role in ChatMessage is a string with a fixed set of values. It's better to define this as a str, Enum for improved type safety and developer experience.

Suggested change
@dataclass
class ChatMessage:
"""Chat message with role support."""
role: str # "system", "user", "assistant"
content: str
name: Optional[str] = None
class ChatRole(str, Enum):
"""Role of the message author."""
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
@dataclass
class ChatMessage:
"""Chat message with role support."""
role: ChatRole
content: str
name: Optional[str] = None



@dataclass
class ChatChoice:
"""Single chat choice."""
index: int
message: dict
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The message field is typed as a dict. For better type safety and consistency, it should be typed as ChatMessage. This requires updating the parsing logic in _parse_completion.

Suggested change
message: dict
message: ChatMessage

finish_reason: Optional[str] = None


@dataclass
class Usage:
"""Token usage."""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0


@dataclass
class ChatCompletion:
"""Native Python completion response."""
id: str
object_type: str = "chat.completion"
created: int = 0
model: str = ""
choices: list[ChatChoice] = field(default_factory=list)
usage: Usage = field(default_factory=Usage)

@property
def first_choice(self) -> str:
if self.choices and self.choices[0].message:
return self.choices[0].message.get("content", "")
return ""
Comment on lines +92 to +95
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

With message being a ChatMessage object, you can now access its content attribute directly for cleaner code.

Suggested change
def first_choice(self) -> str:
if self.choices and self.choices[0].message:
return self.choices[0].message.get("content", "")
return ""
def first_choice(self) -> str:
if self.choices and self.choices[0].message:
return self.choices[0].message.content or ""
return ""


@property
def text(self) -> str:
return self.first_choice

@property
def content(self) -> str:
return self.first_choice


@dataclass
class Model:
"""Model info."""
id: str
object_type: str = "model"
created: Optional[int] = None
owned_by: Optional[str] = None


@dataclass
class ModelList:
"""List of models."""
object_type: str = "list"
data: list[Model] = field(default_factory=list)


# =============================================================================
# Client - Full-featured Python SDK
# =============================================================================

class CliproxyClient:
"""Comprehensive Python SDK - NOT just HTTP wrapper.

Provides native Python classes and functions for cliproxyapi-plusplus.
"""

def __init__(
self,
base_url: str = "http://127.0.0.1:8317",
api_key: Optional[str] = None,
timeout: int = 30,
):
self.base_url = base_url.rstrip("/")
self.api_key = api_key or os.getenv("CLIPROXY_API_KEY", "8317")
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Using a hardcoded default value of "8317" for the API key is a security risk. This value appears to be the default port, which could be confusing and lead to insecure deployments if a user forgets to set a proper key. It's better to default to None if the environment variable is not set. The _request method should then be updated to only add the Authorization header if an API key is present.

Suggested change
self.api_key = api_key or os.getenv("CLIPROXY_API_KEY", "8317")
self.api_key = api_key or os.getenv("CLIPROXY_API_KEY")

self.timeout = timeout
self._client = httpx.Client(timeout=timeout)

# -------------------------------------------------------------------------
# High-level Python methods (not HTTP mapping)
# -------------------------------------------------------------------------

def chat(
self,
messages: list[ChatMessage],
model: str = "claude-3-5-sonnet-20241022",
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Use the DEFAULT_MODEL constant here and in other places where this model name is hardcoded (lines 164, 273) to improve maintainability.

Suggested change
model: str = "claude-3-5-sonnet-20241022",
model: str = DEFAULT_MODEL,

**kwargs
) -> ChatCompletion:
"""Native Python chat - returns ChatCompletion object."""
resp = self.completions_create(
model=model,
messages=[{"role": m.role, "content": m.content} for m in messages],
**kwargs
)
return self._parse_completion(resp)

def complete(
self,
prompt: str,
model: str = "claude-3-5-sonnet-20241022",
system: Optional[str] = None,
) -> str:
"""Simple completion - returns string."""
msgs = []
if system:
msgs.append(ChatMessage(role="system", content=system))
msgs.append(ChatMessage(role="user", content=prompt))

resp = self.chat(msgs, model)
return resp.first_choice

# -------------------------------------------------------------------------
# Mid-level operations
# -------------------------------------------------------------------------

def providers_list(self) -> list[str]:
"""List available providers."""
return [p.value for p in ModelProvider]

def auth_add(self, auth: AuthEntry) -> dict:
"""Add auth entry - native Python."""
return self.management_request("POST", "/v0/management/auth", json=auth.__dict__)
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using __dict__ to serialize a dataclass is not recommended. The dataclasses.asdict() function should be used instead, as it's the official and safer way to convert a dataclass instance to a dictionary, and it correctly handles nested dataclasses.

Suggested change
return self.management_request("POST", "/v0/management/auth", json=auth.__dict__)
return self.management_request("POST", "/v0/management/auth", json=asdict(auth))


def config_update(self, **kwargs) -> dict:
"""Update config with kwargs."""
return self.management_request("PUT", "/v0/management/config", json=kwargs)

def models(self) -> ModelList:
"""List models as ModelList."""
resp = self._request("GET", "/v1/models")
return ModelList(
object_type=resp.get("object", "list"),
data=[Model(**m) for m in resp.get("data", [])]
)

# -------------------------------------------------------------------------
# Low-level HTTP
# -------------------------------------------------------------------------

def completions_create(self, **kwargs) -> dict:
"""Raw OpenAI-compatible /v1/chat/completions."""
return self._request("POST", "/v1/chat/completions", json=kwargs)

def models_list_raw(self) -> dict:
"""List models raw."""
return self._request("GET", "/v1/models")

def management_request(
self,
method: str,
path: str,
**kwargs
) -> dict:
"""Management API."""
return self._request(method, f"/v0/management{path}", **kwargs)

def _request(
self,
method: str,
path: str,
**kwargs
) -> dict:
"""Base HTTP request."""
url = f"{self.base_url}{path}"
headers = {"Authorization": f"Bearer {self.api_key}"}
headers.update(kwargs.pop("headers", {}))
Comment on lines +229 to +230
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

To handle cases where self.api_key might be None, the Authorization header should only be added if a key is present.

Suggested change
headers = {"Authorization": f"Bearer {self.api_key}"}
headers.update(kwargs.pop("headers", {}))
headers = {}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
headers.update(kwargs.pop("headers", {}))


resp = self._client.request(method, url, headers=headers, **kwargs)
resp.raise_for_status()
return resp.json()

def _parse_completion(self, resp: dict) -> ChatCompletion:
"""Parse completion response to Python object."""
choices = [ChatChoice(**c) for c in resp.get("choices", [])]
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To support ChatChoice.message being a ChatMessage object, the parsing logic needs to be updated to explicitly construct ChatMessage objects from the response data.

Suggested change
choices = [ChatChoice(**c) for c in resp.get("choices", [])]
choices = [
ChatChoice(
index=c.get("index", 0),
message=ChatMessage(**c.get("message", {})),
finish_reason=c.get("finish_reason")
) for c in resp.get("choices", [])
]

usage_data = resp.get("usage", {})
usage = Usage(
prompt_tokens=usage_data.get("prompt_tokens", 0),
completion_tokens=usage_data.get("completion_tokens", 0),
total_tokens=usage_data.get("total_tokens", 0)
)
return ChatCompletion(
id=resp.get("id", ""),
object_type=resp.get("object", "chat.completion"),
created=resp.get("created", 0),
model=resp.get("model", ""),
choices=choices,
usage=usage
)

def close(self):
self._client.close()

def __enter__(self):
return self

def __exit__(self, *args):
self.close()


# =============================================================================
# Convenience functions
# =============================================================================

def client(**kwargs) -> CliproxyClient:
"""Create client - shortcut."""
return CliproxyClient(**kwargs)


def chat(prompt: str, model: str = "claude-3-5-sonnet-20241022", **kwargs) -> str:
"""One-shot chat - returns string."""
with CliproxyClient() as c:
return c.complete(prompt, model, **kwargs)
Loading