mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(mcp): add auth, kms, and platform MCP tools
Add 3 new hanzo-tools-* packages exposing Hanzo platform services via MCP: - hanzo-tools-auth: HanzoSession singleton + LoginTool (status/whoami/logout/refresh) - hanzo-tools-kms: KMSTool (list/get/set/delete/inject secrets via KMS) - hanzo-tools-platform: PlatformTool (PaaS deployments, IAM users/orgs, cloud services) Wire into hanzo-mcp as dependencies and register in entrypoint loader.
This commit is contained in:
@@ -56,6 +56,9 @@ PACKAGE_TOOL_PREFIXES: dict[str, list[str]] = {
|
||||
"config": ["config", "mode"],
|
||||
"mcp_tools": ["mcp"],
|
||||
"api": ["api"],
|
||||
"auth": ["auth"],
|
||||
"kms": ["kms"],
|
||||
"platform": ["platform"],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -40,6 +40,9 @@ dependencies = [
|
||||
"hanzo-tools-llm>=0.1.0",
|
||||
"hanzo-tools-agent>=0.3.1",
|
||||
"hanzo-tools-api>=0.3.1",
|
||||
"hanzo-tools-auth>=0.1.0",
|
||||
"hanzo-tools-kms>=0.1.0",
|
||||
"hanzo-tools-platform>=0.1.0",
|
||||
"hanzo-persona>=1.0.0",
|
||||
"uvloop>=0.22.1",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
# hanzo-tools-auth
|
||||
|
||||
Authentication bridge for Hanzo MCP platform tools.
|
||||
@@ -0,0 +1,17 @@
|
||||
"""Hanzo Auth Tools — authentication bridge for MCP platform tools.
|
||||
|
||||
Provides the HanzoSession singleton for shared auth state, and
|
||||
the LoginTool MCP tool for auth management.
|
||||
"""
|
||||
|
||||
from .login_tool import LoginTool
|
||||
from .session import HanzoSession
|
||||
|
||||
# Tools list for entry point discovery
|
||||
TOOLS = [LoginTool]
|
||||
|
||||
__all__ = [
|
||||
"LoginTool",
|
||||
"HanzoSession",
|
||||
"TOOLS",
|
||||
]
|
||||
@@ -0,0 +1,170 @@
|
||||
"""MCP tool for Hanzo authentication management.
|
||||
|
||||
Provides status, login (browser flow), logout, and whoami actions
|
||||
accessible via Claude Code or any MCP client.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Annotated, Any, final
|
||||
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
from pydantic import Field
|
||||
|
||||
from hanzo_tools.core.base import BaseTool
|
||||
|
||||
from .session import HanzoSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DESCRIPTION = """Hanzo authentication management.
|
||||
|
||||
Manage your Hanzo platform authentication. Check auth status, view current user
|
||||
info, or logout.
|
||||
|
||||
Actions:
|
||||
- status: Show current authentication state and accessible services
|
||||
- whoami: Show current user info from IAM token
|
||||
- logout: Clear stored credentials
|
||||
- refresh: Refresh an expired token
|
||||
"""
|
||||
|
||||
|
||||
@final
|
||||
class LoginTool(BaseTool):
|
||||
"""MCP tool for authentication operations."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "auth"
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return DESCRIPTION
|
||||
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "status",
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
session = HanzoSession.get()
|
||||
|
||||
if action == "status":
|
||||
return await self._status(session)
|
||||
elif action == "whoami":
|
||||
return await self._whoami(session)
|
||||
elif action == "logout":
|
||||
return await self._logout(session)
|
||||
elif action == "refresh":
|
||||
return await self._refresh(session)
|
||||
else:
|
||||
return json.dumps({"error": f"Unknown action: {action}. Use: status, whoami, logout, refresh"})
|
||||
|
||||
async def _status(self, session: HanzoSession) -> str:
|
||||
info = session.get_token_info()
|
||||
|
||||
if not info.get("authenticated"):
|
||||
return json.dumps({
|
||||
"authenticated": False,
|
||||
"message": "Not authenticated. Run 'hanzo login' in your terminal to authenticate.",
|
||||
"services": {
|
||||
"iam": False,
|
||||
"kms": _check_kms_env(),
|
||||
"paas": False,
|
||||
},
|
||||
}, indent=2)
|
||||
|
||||
services = {
|
||||
"iam": True,
|
||||
"kms": _check_kms_env(),
|
||||
"paas": info.get("authenticated", False),
|
||||
}
|
||||
|
||||
return json.dumps({
|
||||
"authenticated": True,
|
||||
"source": info.get("source", "unknown"),
|
||||
"organization": info.get("organization"),
|
||||
"server_url": info.get("server_url"),
|
||||
"expired": info.get("expired", False),
|
||||
"services": services,
|
||||
}, indent=2)
|
||||
|
||||
async def _whoami(self, session: HanzoSession) -> str:
|
||||
if not session.is_authenticated():
|
||||
return json.dumps({"error": "Not authenticated. Run 'hanzo login' first."})
|
||||
|
||||
try:
|
||||
iam_client = session.get_iam_client()
|
||||
token = session.get_iam_token()
|
||||
if token:
|
||||
# Try to decode JWT claims (without verification for display)
|
||||
try:
|
||||
import jwt
|
||||
|
||||
claims = jwt.decode(token, options={"verify_signature": False})
|
||||
return json.dumps({
|
||||
"sub": claims.get("sub"),
|
||||
"name": claims.get("name"),
|
||||
"email": claims.get("email"),
|
||||
"organization": claims.get("owner"),
|
||||
"iss": claims.get("iss"),
|
||||
}, indent=2)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback to stored token info
|
||||
info = session.get_token_info()
|
||||
return json.dumps({
|
||||
"organization": info.get("organization"),
|
||||
"server_url": info.get("server_url"),
|
||||
"source": info.get("source"),
|
||||
}, indent=2)
|
||||
|
||||
except Exception as e:
|
||||
return json.dumps({"error": f"Failed to get user info: {e}"})
|
||||
|
||||
async def _logout(self, session: HanzoSession) -> str:
|
||||
session.logout()
|
||||
session.close()
|
||||
HanzoSession.reset()
|
||||
return json.dumps({"message": "Logged out. Cleared stored credentials."})
|
||||
|
||||
async def _refresh(self, session: HanzoSession) -> str:
|
||||
if not session.is_authenticated():
|
||||
return json.dumps({"error": "Not authenticated. Run 'hanzo login' first."})
|
||||
|
||||
if session.refresh_token():
|
||||
return json.dumps({"message": "Token refreshed successfully."})
|
||||
else:
|
||||
return json.dumps({"error": "Token refresh failed. Run 'hanzo login' again."})
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register auth tool with explicit parameters."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool(
|
||||
name="auth",
|
||||
description=DESCRIPTION,
|
||||
)
|
||||
async def auth(
|
||||
action: Annotated[
|
||||
str,
|
||||
Field(
|
||||
description="Action: status (check auth state), whoami (current user), logout, refresh",
|
||||
default="status",
|
||||
),
|
||||
] = "status",
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
return await tool_instance.call(ctx, action=action)
|
||||
|
||||
|
||||
def _check_kms_env() -> bool:
|
||||
"""Check if KMS credentials are available."""
|
||||
import os
|
||||
|
||||
return bool(os.getenv("HANZO_KMS_CLIENT_ID") and os.getenv("HANZO_KMS_CLIENT_SECRET"))
|
||||
@@ -0,0 +1,393 @@
|
||||
"""Hanzo authentication session — shared auth bridge for MCP platform tools.
|
||||
|
||||
Loads IAM tokens from disk or environment, auto-refreshes expired tokens,
|
||||
and provides authenticated service clients (KMS, PaaS, IAM).
|
||||
|
||||
Token resolution order:
|
||||
1. HANZO_AUTH_TOKEN env var (explicit override)
|
||||
2. HANZO_API_KEY env var (API key auth)
|
||||
3. ~/.hanzo/auth/token.json (from `hanzo login`)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TOKEN_DIR = Path.home() / ".hanzo" / "auth"
|
||||
TOKEN_FILE = TOKEN_DIR / "token.json"
|
||||
|
||||
# IAM defaults (same as hanzo-cli)
|
||||
DEFAULT_IAM_URL = "https://hanzo.id"
|
||||
DEFAULT_ORG = "hanzo"
|
||||
DEFAULT_APP = "app-hanzo"
|
||||
DEFAULT_CLIENT_ID = "hanzo-app-client-id"
|
||||
|
||||
|
||||
def _env(name: str) -> str:
|
||||
"""Read env var, accepting both IAM_ and HANZO_IAM_ prefixes."""
|
||||
return os.getenv(name) or os.getenv(f"HANZO_{name}") or ""
|
||||
|
||||
|
||||
class HanzoSession:
|
||||
"""Singleton session providing authenticated clients to platform tools."""
|
||||
|
||||
_instance: HanzoSession | None = None
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._token_data: dict[str, Any] | None = None
|
||||
self._iam_client: Any | None = None
|
||||
self._kms_client: Any | None = None
|
||||
|
||||
@classmethod
|
||||
def get(cls) -> HanzoSession:
|
||||
"""Get or create the singleton session."""
|
||||
if cls._instance is None:
|
||||
cls._instance = cls()
|
||||
return cls._instance
|
||||
|
||||
@classmethod
|
||||
def reset(cls) -> None:
|
||||
"""Reset the singleton (for testing)."""
|
||||
if cls._instance:
|
||||
cls._instance.close()
|
||||
cls._instance = None
|
||||
|
||||
# -- Token loading -------------------------------------------------------
|
||||
|
||||
def _load_token_from_disk(self) -> dict[str, Any] | None:
|
||||
"""Load stored token from ~/.hanzo/auth/token.json."""
|
||||
if not TOKEN_FILE.exists():
|
||||
return None
|
||||
try:
|
||||
return json.loads(TOKEN_FILE.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return None
|
||||
|
||||
def _save_token(self, data: dict[str, Any]) -> None:
|
||||
"""Save token data to disk."""
|
||||
TOKEN_DIR.mkdir(parents=True, exist_ok=True)
|
||||
TOKEN_FILE.write_text(json.dumps(data, indent=2))
|
||||
TOKEN_FILE.chmod(0o600)
|
||||
|
||||
def load_token(self) -> dict[str, Any] | None:
|
||||
"""Load token using the resolution chain.
|
||||
|
||||
Returns token data dict or None if not authenticated.
|
||||
"""
|
||||
if self._token_data:
|
||||
return self._token_data
|
||||
|
||||
# 1. Explicit token override
|
||||
auth_token = os.getenv("HANZO_AUTH_TOKEN")
|
||||
if auth_token:
|
||||
self._token_data = {
|
||||
"access_token": auth_token,
|
||||
"source": "env:HANZO_AUTH_TOKEN",
|
||||
}
|
||||
return self._token_data
|
||||
|
||||
# 2. API key
|
||||
api_key = os.getenv("HANZO_API_KEY")
|
||||
if api_key:
|
||||
self._token_data = {
|
||||
"access_token": api_key,
|
||||
"source": "env:HANZO_API_KEY",
|
||||
}
|
||||
return self._token_data
|
||||
|
||||
# 3. Stored token from `hanzo login`
|
||||
token_data = self._load_token_from_disk()
|
||||
if token_data and token_data.get("access_token"):
|
||||
token_data["source"] = "disk:~/.hanzo/auth/token.json"
|
||||
self._token_data = token_data
|
||||
return self._token_data
|
||||
|
||||
return None
|
||||
|
||||
# -- Token state ---------------------------------------------------------
|
||||
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Check if we have a valid token."""
|
||||
return self.load_token() is not None
|
||||
|
||||
def get_iam_token(self) -> str | None:
|
||||
"""Get the current IAM access token."""
|
||||
token_data = self.load_token()
|
||||
if token_data:
|
||||
return token_data.get("access_token")
|
||||
return None
|
||||
|
||||
def get_token_info(self) -> dict[str, Any]:
|
||||
"""Get info about the current auth state."""
|
||||
token_data = self.load_token()
|
||||
if not token_data:
|
||||
return {"authenticated": False}
|
||||
|
||||
info: dict[str, Any] = {
|
||||
"authenticated": True,
|
||||
"source": token_data.get("source", "unknown"),
|
||||
}
|
||||
|
||||
# Check expiry
|
||||
login_time = token_data.get("login_time", 0)
|
||||
expires_in = token_data.get("expires_in", 0)
|
||||
if login_time and expires_in:
|
||||
expires_at = login_time + expires_in
|
||||
info["expires_at"] = expires_at
|
||||
info["expired"] = time.time() > expires_at
|
||||
|
||||
# Add org/app info if available
|
||||
if token_data.get("organization"):
|
||||
info["organization"] = token_data["organization"]
|
||||
if token_data.get("application"):
|
||||
info["application"] = token_data["application"]
|
||||
if token_data.get("server_url"):
|
||||
info["server_url"] = token_data["server_url"]
|
||||
|
||||
return info
|
||||
|
||||
# -- Token refresh -------------------------------------------------------
|
||||
|
||||
def refresh_token(self) -> bool:
|
||||
"""Attempt to refresh an expired token.
|
||||
|
||||
Returns True if refresh succeeded.
|
||||
"""
|
||||
token_data = self._load_token_from_disk()
|
||||
if not token_data or not token_data.get("refresh_token"):
|
||||
return False
|
||||
|
||||
try:
|
||||
from hanzo_iam import IAMClient, IAMConfig
|
||||
|
||||
config = IAMConfig(
|
||||
server_url=token_data.get("server_url", DEFAULT_IAM_URL),
|
||||
client_id=token_data.get("client_id", DEFAULT_CLIENT_ID),
|
||||
client_secret="",
|
||||
organization=token_data.get("organization", DEFAULT_ORG),
|
||||
application=token_data.get("application", DEFAULT_APP),
|
||||
)
|
||||
client = IAMClient(config=config)
|
||||
|
||||
tokens = client.refresh_token(token_data["refresh_token"])
|
||||
client.close()
|
||||
|
||||
new_data = {
|
||||
**token_data,
|
||||
"access_token": tokens.access_token,
|
||||
"refresh_token": tokens.refresh_token or token_data["refresh_token"],
|
||||
"id_token": getattr(tokens, "id_token", ""),
|
||||
"expires_in": tokens.expires_in,
|
||||
"login_time": int(time.time()),
|
||||
}
|
||||
|
||||
self._save_token(new_data)
|
||||
self._token_data = new_data
|
||||
self._token_data["source"] = "disk:~/.hanzo/auth/token.json"
|
||||
logger.info("Token refreshed successfully")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Token refresh failed: {e}")
|
||||
return False
|
||||
|
||||
# -- Service clients -----------------------------------------------------
|
||||
|
||||
def get_iam_client(self) -> Any:
|
||||
"""Get an authenticated IAMClient."""
|
||||
if self._iam_client:
|
||||
return self._iam_client
|
||||
|
||||
from hanzo_iam import IAMClient, IAMConfig
|
||||
|
||||
token_data = self.load_token()
|
||||
if not token_data:
|
||||
raise RuntimeError("Not authenticated. Run 'hanzo login' first.")
|
||||
|
||||
# If we have M2M credentials
|
||||
client_id = _env("IAM_CLIENT_ID")
|
||||
client_secret = _env("IAM_CLIENT_SECRET")
|
||||
|
||||
if client_id and client_secret:
|
||||
config = IAMConfig(
|
||||
server_url=_env("IAM_URL") or DEFAULT_IAM_URL,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
organization=_env("IAM_ORG") or DEFAULT_ORG,
|
||||
application=_env("IAM_APP") or DEFAULT_APP,
|
||||
)
|
||||
self._iam_client = IAMClient(config=config)
|
||||
else:
|
||||
config = IAMConfig(
|
||||
server_url=token_data.get("server_url", DEFAULT_IAM_URL),
|
||||
client_id=token_data.get("client_id", ""),
|
||||
client_secret="",
|
||||
organization=token_data.get("organization", DEFAULT_ORG),
|
||||
application=token_data.get("application", DEFAULT_APP),
|
||||
)
|
||||
self._iam_client = IAMClient(
|
||||
config=config,
|
||||
bearer_token=token_data["access_token"],
|
||||
)
|
||||
|
||||
return self._iam_client
|
||||
|
||||
def get_kms_client(self) -> Any:
|
||||
"""Get an authenticated KMSClient."""
|
||||
if self._kms_client:
|
||||
return self._kms_client
|
||||
|
||||
from hanzo_kms import ClientSettings, KMSClient
|
||||
|
||||
kms_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
|
||||
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
|
||||
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
|
||||
|
||||
if client_id and client_secret:
|
||||
from hanzo_kms import AuthenticationOptions, UniversalAuthMethod
|
||||
|
||||
settings = ClientSettings(
|
||||
site_url=kms_url,
|
||||
auth=AuthenticationOptions(
|
||||
universal_auth=UniversalAuthMethod(
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
)
|
||||
),
|
||||
)
|
||||
self._kms_client = KMSClient(settings=settings)
|
||||
else:
|
||||
# Fall back to default env-based construction
|
||||
self._kms_client = KMSClient()
|
||||
|
||||
return self._kms_client
|
||||
|
||||
def get_paas_client(self) -> Any:
|
||||
"""Get an authenticated PaaS client via IAM token exchange."""
|
||||
import httpx
|
||||
|
||||
token_data = self.load_token()
|
||||
if not token_data:
|
||||
raise RuntimeError("Not authenticated. Run 'hanzo login' first.")
|
||||
|
||||
base_url = os.getenv("HANZO_PAAS_URL", "https://platform.hanzo.ai").rstrip("/")
|
||||
iam_token = token_data["access_token"]
|
||||
|
||||
# Check for cached PaaS session
|
||||
session_file = Path.home() / ".hanzo" / "paas" / "session.json"
|
||||
if session_file.exists():
|
||||
try:
|
||||
session = json.loads(session_file.read_text())
|
||||
if session.get("at"):
|
||||
# Validate cached session
|
||||
with httpx.Client(base_url=base_url, timeout=10.0) as tmp:
|
||||
resp = tmp.get(
|
||||
"/v1/org",
|
||||
headers={"Authorization": session["at"]},
|
||||
)
|
||||
if resp.status_code != 401:
|
||||
return _PaaSClientWrapper(base_url, session["at"], session.get("rt"))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Exchange IAM token for PaaS session
|
||||
with httpx.Client(base_url=base_url, timeout=30.0) as tmp:
|
||||
resp = tmp.post(
|
||||
"/v1/auth/login",
|
||||
json={"provider": "hanzo", "accessToken": iam_token},
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
if resp.status_code == 401:
|
||||
raise RuntimeError("IAM token rejected by PaaS. Try 'hanzo login' again.")
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
at = data.get("at", "")
|
||||
rt = data.get("rt", "")
|
||||
if not at:
|
||||
raise RuntimeError("PaaS login succeeded but no session token returned.")
|
||||
|
||||
# Cache session
|
||||
session_dir = Path.home() / ".hanzo" / "paas"
|
||||
session_dir.mkdir(parents=True, exist_ok=True)
|
||||
session_file = session_dir / "session.json"
|
||||
session_file.write_text(json.dumps({"at": at, "rt": rt, "login_time": int(time.time())}, indent=2))
|
||||
session_file.chmod(0o600)
|
||||
|
||||
return _PaaSClientWrapper(base_url, at, rt)
|
||||
|
||||
# -- Lifecycle -----------------------------------------------------------
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close all held clients."""
|
||||
if self._iam_client and hasattr(self._iam_client, "close"):
|
||||
self._iam_client.close()
|
||||
if self._kms_client and hasattr(self._kms_client, "close"):
|
||||
self._kms_client.close()
|
||||
self._iam_client = None
|
||||
self._kms_client = None
|
||||
self._token_data = None
|
||||
|
||||
# -- Logout --------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def logout() -> None:
|
||||
"""Clear stored credentials."""
|
||||
if TOKEN_FILE.exists():
|
||||
TOKEN_FILE.unlink()
|
||||
session_file = Path.home() / ".hanzo" / "paas" / "session.json"
|
||||
if session_file.exists():
|
||||
session_file.unlink()
|
||||
|
||||
|
||||
class _PaaSClientWrapper:
|
||||
"""Lightweight async-friendly wrapper over PaaS REST API."""
|
||||
|
||||
def __init__(self, base_url: str, access_token: str, refresh_token: str | None = None):
|
||||
self.base_url = base_url
|
||||
self._at = access_token
|
||||
self._rt = refresh_token
|
||||
|
||||
def _headers(self) -> dict[str, str]:
|
||||
headers = {"Content-Type": "application/json", "User-Agent": "hanzo-mcp/0.1"}
|
||||
if self._at:
|
||||
headers["Authorization"] = self._at
|
||||
if self._rt:
|
||||
headers["Refresh-Token"] = self._rt
|
||||
return headers
|
||||
|
||||
def request(self, method: str, path: str, **kwargs: Any) -> Any:
|
||||
"""Make an HTTP request to PaaS API."""
|
||||
import httpx
|
||||
|
||||
with httpx.Client(base_url=self.base_url, timeout=30.0) as client:
|
||||
resp = client.request(method, path, headers=self._headers(), **kwargs)
|
||||
if resp.status_code >= 400:
|
||||
try:
|
||||
err = resp.json()
|
||||
msg = err.get("error", resp.text)
|
||||
except Exception:
|
||||
msg = resp.text
|
||||
raise RuntimeError(f"PaaS error {resp.status_code}: {msg}")
|
||||
if not resp.content or resp.status_code == 204:
|
||||
return {}
|
||||
return resp.json()
|
||||
|
||||
def get(self, path: str) -> Any:
|
||||
return self.request("GET", path)
|
||||
|
||||
def post(self, path: str, **kwargs: Any) -> Any:
|
||||
return self.request("POST", path, **kwargs)
|
||||
|
||||
def put(self, path: str, **kwargs: Any) -> Any:
|
||||
return self.request("PUT", path, **kwargs)
|
||||
|
||||
def delete(self, path: str) -> Any:
|
||||
return self.request("DELETE", path)
|
||||
@@ -0,0 +1,33 @@
|
||||
[project]
|
||||
name = "hanzo-tools-auth"
|
||||
version = "0.1.0"
|
||||
description = "Authentication bridge for Hanzo MCP platform tools — session management, token refresh, service client factory"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
|
||||
keywords = ["hanzo", "mcp", "auth", "iam", "tools"]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"hanzo-tools-core>=0.1.0",
|
||||
"hanzo-iam>=0.1.0",
|
||||
"httpx>=0.27.0",
|
||||
"pyjwt>=2.8.0",
|
||||
]
|
||||
|
||||
[project.entry-points."hanzo.tools"]
|
||||
auth = "hanzo_tools.auth:TOOLS"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["hanzo_tools"]
|
||||
@@ -0,0 +1,3 @@
|
||||
# hanzo-tools-kms
|
||||
|
||||
KMS secret management MCP tool for Hanzo platform.
|
||||
@@ -0,0 +1,7 @@
|
||||
"""Hanzo KMS Tools — secret management via MCP."""
|
||||
|
||||
from .kms_tool import KMSTool
|
||||
|
||||
TOOLS = [KMSTool]
|
||||
|
||||
__all__ = ["KMSTool", "TOOLS"]
|
||||
@@ -0,0 +1,296 @@
|
||||
"""MCP tool for Hanzo KMS secret management.
|
||||
|
||||
Wraps the hanzo-kms client to provide secret CRUD operations
|
||||
via MCP. Auth uses HANZO_KMS_CLIENT_ID / HANZO_KMS_CLIENT_SECRET
|
||||
environment variables (same pattern as hanzo-cli).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Annotated, Any, final
|
||||
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
from pydantic import Field
|
||||
|
||||
from hanzo_tools.core.base import BaseTool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DESCRIPTION = """Manage secrets via Hanzo KMS.
|
||||
|
||||
Requires HANZO_KMS_CLIENT_ID and HANZO_KMS_CLIENT_SECRET environment variables.
|
||||
|
||||
Actions:
|
||||
- list: List all secrets in a project/environment (values masked)
|
||||
- get: Get a single secret value
|
||||
- set: Create or update a secret
|
||||
- delete: Remove a secret
|
||||
- inject: Output secrets as export/dotenv/json format
|
||||
"""
|
||||
|
||||
|
||||
def _get_kms_client() -> Any:
|
||||
"""Build a KMS client from environment."""
|
||||
from hanzo_kms import ClientSettings, KMSClient
|
||||
|
||||
kms_url = os.getenv("HANZO_KMS_URL", "https://kms.hanzo.ai")
|
||||
client_id = os.getenv("HANZO_KMS_CLIENT_ID", "")
|
||||
client_secret = os.getenv("HANZO_KMS_CLIENT_SECRET", "")
|
||||
|
||||
if client_id and client_secret:
|
||||
from hanzo_kms import AuthenticationOptions, UniversalAuthMethod
|
||||
|
||||
settings = ClientSettings(
|
||||
site_url=kms_url,
|
||||
auth=AuthenticationOptions(
|
||||
universal_auth=UniversalAuthMethod(
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
)
|
||||
),
|
||||
)
|
||||
return KMSClient(settings=settings)
|
||||
|
||||
# Fall back to default env-based construction
|
||||
return KMSClient()
|
||||
|
||||
|
||||
@final
|
||||
class KMSTool(BaseTool):
|
||||
"""MCP tool for KMS secret management."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "kms"
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return DESCRIPTION
|
||||
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "list",
|
||||
project: str | None = None,
|
||||
environment: str | None = None,
|
||||
secret_name: str | None = None,
|
||||
secret_value: str | None = None,
|
||||
path: str = "/",
|
||||
format: str = "export",
|
||||
reveal: bool = False,
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
if action == "list":
|
||||
return await self._list(project, environment, path)
|
||||
elif action == "get":
|
||||
return await self._get(project, environment, secret_name, path, reveal)
|
||||
elif action == "set":
|
||||
return await self._set(project, environment, secret_name, secret_value, path)
|
||||
elif action == "delete":
|
||||
return await self._delete(project, environment, secret_name, path)
|
||||
elif action == "inject":
|
||||
return await self._inject(project, environment, path, format)
|
||||
else:
|
||||
return json.dumps({"error": f"Unknown action: {action}. Use: list, get, set, delete, inject"})
|
||||
|
||||
async def _list(self, project: str | None, env: str | None, path: str) -> str:
|
||||
if not project or not env:
|
||||
return json.dumps({"error": "Required: project and environment"})
|
||||
|
||||
client = _get_kms_client()
|
||||
try:
|
||||
secrets = client.list_secrets(project_id=project, environment=env, path=path)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
if not secrets:
|
||||
return json.dumps({"message": f"No secrets found in {project}/{env}", "secrets": []})
|
||||
|
||||
rows = []
|
||||
for s in secrets:
|
||||
val = s.secret_value
|
||||
masked = f"{val[:4]}***" if len(val) > 4 else "***"
|
||||
rows.append({
|
||||
"key": s.secret_key,
|
||||
"value": masked,
|
||||
"version": s.version,
|
||||
"updated": str(s.updated_at)[:19] if s.updated_at else None,
|
||||
})
|
||||
|
||||
return json.dumps({
|
||||
"project": project,
|
||||
"environment": env,
|
||||
"path": path,
|
||||
"count": len(rows),
|
||||
"secrets": rows,
|
||||
}, indent=2)
|
||||
|
||||
async def _get(
|
||||
self, project: str | None, env: str | None, name: str | None, path: str, reveal: bool
|
||||
) -> str:
|
||||
if not project or not env or not name:
|
||||
return json.dumps({"error": "Required: project, environment, and secret_name"})
|
||||
|
||||
client = _get_kms_client()
|
||||
try:
|
||||
secret = client.get_secret(
|
||||
project_id=project, environment=env, secret_name=name, path=path
|
||||
)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
val = secret.secret_value
|
||||
if not reveal:
|
||||
val = f"{val[:4]}***" if len(val) > 4 else "***"
|
||||
|
||||
return json.dumps({
|
||||
"key": secret.secret_key,
|
||||
"value": val,
|
||||
"version": secret.version,
|
||||
"type": secret.type,
|
||||
"environment": secret.environment,
|
||||
"comment": secret.secret_comment or None,
|
||||
"revealed": reveal,
|
||||
}, indent=2)
|
||||
|
||||
async def _set(
|
||||
self, project: str | None, env: str | None, name: str | None, value: str | None, path: str
|
||||
) -> str:
|
||||
if not project or not env or not name or not value:
|
||||
return json.dumps({"error": "Required: project, environment, secret_name, and secret_value"})
|
||||
|
||||
client = _get_kms_client()
|
||||
try:
|
||||
# Try update first, create if it doesn't exist
|
||||
try:
|
||||
secret = client.update_secret(
|
||||
project_id=project,
|
||||
environment=env,
|
||||
secret_name=name,
|
||||
secret_value=value,
|
||||
)
|
||||
return json.dumps({
|
||||
"action": "updated",
|
||||
"key": name,
|
||||
"version": secret.version,
|
||||
})
|
||||
except Exception:
|
||||
secret = client.create_secret(
|
||||
project_id=project,
|
||||
environment=env,
|
||||
secret_name=name,
|
||||
secret_value=value,
|
||||
)
|
||||
return json.dumps({
|
||||
"action": "created",
|
||||
"key": name,
|
||||
})
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
async def _delete(
|
||||
self, project: str | None, env: str | None, name: str | None, path: str
|
||||
) -> str:
|
||||
if not project or not env or not name:
|
||||
return json.dumps({"error": "Required: project, environment, and secret_name"})
|
||||
|
||||
client = _get_kms_client()
|
||||
try:
|
||||
client.delete_secret(
|
||||
project_id=project,
|
||||
environment=env,
|
||||
secret_name=name,
|
||||
path=path,
|
||||
)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
return json.dumps({"action": "deleted", "key": name})
|
||||
|
||||
async def _inject(self, project: str | None, env: str | None, path: str, fmt: str) -> str:
|
||||
if not project or not env:
|
||||
return json.dumps({"error": "Required: project and environment"})
|
||||
|
||||
client = _get_kms_client()
|
||||
try:
|
||||
secrets = client.list_secrets(project_id=project, environment=env, path=path)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
if not secrets:
|
||||
return json.dumps({"message": "No secrets found", "output": ""})
|
||||
|
||||
if fmt == "json":
|
||||
data = {s.secret_key: s.secret_value for s in secrets}
|
||||
return json.dumps(data, indent=2)
|
||||
elif fmt == "dotenv":
|
||||
lines = []
|
||||
for s in secrets:
|
||||
val = s.secret_value.replace('"', '\\"')
|
||||
lines.append(f'{s.secret_key}="{val}"')
|
||||
return "\n".join(lines)
|
||||
else: # export
|
||||
lines = []
|
||||
for s in secrets:
|
||||
val = s.secret_value.replace("'", "'\\''")
|
||||
lines.append(f"export {s.secret_key}='{val}'")
|
||||
return "\n".join(lines)
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register KMS tool with explicit parameters."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool(
|
||||
name="kms",
|
||||
description=DESCRIPTION,
|
||||
)
|
||||
async def kms(
|
||||
action: Annotated[
|
||||
str,
|
||||
Field(description="Action: list, get, set, delete, inject"),
|
||||
] = "list",
|
||||
project: Annotated[
|
||||
str | None,
|
||||
Field(description="KMS project ID or slug"),
|
||||
] = None,
|
||||
environment: Annotated[
|
||||
str | None,
|
||||
Field(description="Environment: dev, staging, production"),
|
||||
] = None,
|
||||
secret_name: Annotated[
|
||||
str | None,
|
||||
Field(description="Secret key name (for get/set/delete)"),
|
||||
] = None,
|
||||
secret_value: Annotated[
|
||||
str | None,
|
||||
Field(description="Secret value (for set)"),
|
||||
] = None,
|
||||
path: Annotated[
|
||||
str,
|
||||
Field(description="Secret path prefix (default: /)"),
|
||||
] = "/",
|
||||
format: Annotated[
|
||||
str,
|
||||
Field(description="Output format for inject: export, dotenv, json"),
|
||||
] = "export",
|
||||
reveal: Annotated[
|
||||
bool,
|
||||
Field(description="Show full secret value (default: masked)"),
|
||||
] = False,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
return await tool_instance.call(
|
||||
ctx,
|
||||
action=action,
|
||||
project=project,
|
||||
environment=environment,
|
||||
secret_name=secret_name,
|
||||
secret_value=secret_value,
|
||||
path=path,
|
||||
format=format,
|
||||
reveal=reveal,
|
||||
)
|
||||
@@ -0,0 +1,31 @@
|
||||
[project]
|
||||
name = "hanzo-tools-kms"
|
||||
version = "0.1.0"
|
||||
description = "KMS secret management MCP tool — list, get, set, delete, inject secrets via Hanzo KMS"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
|
||||
keywords = ["hanzo", "mcp", "kms", "secrets", "tools"]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"hanzo-tools-core>=0.1.0",
|
||||
"hanzo-kms>=0.1.0",
|
||||
]
|
||||
|
||||
[project.entry-points."hanzo.tools"]
|
||||
kms = "hanzo_tools.kms:TOOLS"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["hanzo_tools"]
|
||||
@@ -0,0 +1,3 @@
|
||||
# hanzo-tools-platform
|
||||
|
||||
Hanzo platform MCP tool — PaaS deployments, IAM, and cloud services.
|
||||
@@ -0,0 +1,7 @@
|
||||
"""Hanzo Platform Tools — PaaS, IAM, and cloud services via MCP."""
|
||||
|
||||
from .platform_tool import PlatformTool
|
||||
|
||||
TOOLS = [PlatformTool]
|
||||
|
||||
__all__ = ["PlatformTool", "TOOLS"]
|
||||
@@ -0,0 +1,314 @@
|
||||
"""MCP tool for Hanzo platform services — PaaS, IAM, and cloud.
|
||||
|
||||
Combines PaaS deployments, IAM user/org management, and cloud services
|
||||
into a single MCP tool to avoid tool proliferation.
|
||||
|
||||
Auth: Uses HanzoSession from hanzo-tools-auth to exchange IAM tokens
|
||||
for PaaS sessions and access IAM APIs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Annotated, Any, final
|
||||
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
from pydantic import Field
|
||||
|
||||
from hanzo_tools.core.base import BaseTool
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DESCRIPTION = """Hanzo platform management — PaaS deployments, IAM, and cloud services.
|
||||
|
||||
Requires authentication via `hanzo login` (stored at ~/.hanzo/auth/token.json).
|
||||
|
||||
PaaS actions:
|
||||
- deployments: List containers/deployments in a project environment
|
||||
- deploy: Get details of a specific deployment
|
||||
- logs: View container logs
|
||||
- redeploy: Trigger a rolling restart of a container
|
||||
- env: List environments for a project
|
||||
|
||||
IAM actions:
|
||||
- whoami: Current user info
|
||||
- users: List users in the organization
|
||||
- orgs: List organizations
|
||||
- roles: List roles
|
||||
|
||||
Cloud actions:
|
||||
- services: List managed services (cluster info)
|
||||
- projects: List projects in an organization
|
||||
"""
|
||||
|
||||
|
||||
def _get_session():
|
||||
"""Get HanzoSession singleton."""
|
||||
from hanzo_tools.auth.session import HanzoSession
|
||||
return HanzoSession.get()
|
||||
|
||||
|
||||
@final
|
||||
class PlatformTool(BaseTool):
|
||||
"""MCP tool for Hanzo platform operations."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "platform"
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return DESCRIPTION
|
||||
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "whoami",
|
||||
org: str | None = None,
|
||||
project: str | None = None,
|
||||
environment: str | None = None,
|
||||
container: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
try:
|
||||
# IAM actions
|
||||
if action == "whoami":
|
||||
return await self._whoami()
|
||||
elif action == "users":
|
||||
return await self._users()
|
||||
elif action == "orgs":
|
||||
return await self._orgs()
|
||||
elif action == "roles":
|
||||
return await self._roles()
|
||||
# PaaS actions
|
||||
elif action == "projects":
|
||||
return await self._projects(org)
|
||||
elif action == "env":
|
||||
return await self._envs(org, project)
|
||||
elif action == "deployments":
|
||||
return await self._deployments(org, project, environment)
|
||||
elif action == "deploy":
|
||||
return await self._deploy_detail(org, project, environment, container)
|
||||
elif action == "logs":
|
||||
return await self._logs(org, project, environment, container)
|
||||
elif action == "redeploy":
|
||||
return await self._redeploy(org, project, environment, container)
|
||||
# Cloud actions
|
||||
elif action == "services":
|
||||
return await self._services()
|
||||
else:
|
||||
return json.dumps({
|
||||
"error": f"Unknown action: {action}",
|
||||
"available": [
|
||||
"whoami", "users", "orgs", "roles",
|
||||
"projects", "env", "deployments", "deploy", "logs", "redeploy",
|
||||
"services",
|
||||
],
|
||||
})
|
||||
except RuntimeError as e:
|
||||
return json.dumps({"error": str(e)})
|
||||
except Exception as e:
|
||||
logger.exception(f"Platform tool error: {e}")
|
||||
return json.dumps({"error": f"Platform error: {e}"})
|
||||
|
||||
# -- IAM actions ---------------------------------------------------------
|
||||
|
||||
async def _whoami(self) -> str:
|
||||
session = _get_session()
|
||||
if not session.is_authenticated():
|
||||
return json.dumps({"error": "Not authenticated. Run 'hanzo login' first."})
|
||||
|
||||
token = session.get_iam_token()
|
||||
if token:
|
||||
try:
|
||||
import jwt
|
||||
|
||||
claims = jwt.decode(token, options={"verify_signature": False})
|
||||
return json.dumps({
|
||||
"sub": claims.get("sub"),
|
||||
"name": claims.get("name"),
|
||||
"email": claims.get("email"),
|
||||
"organization": claims.get("owner"),
|
||||
"iss": claims.get("iss"),
|
||||
}, indent=2)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
info = session.get_token_info()
|
||||
return json.dumps(info, indent=2)
|
||||
|
||||
async def _users(self) -> str:
|
||||
session = _get_session()
|
||||
iam = session.get_iam_client()
|
||||
users = iam.get_users()
|
||||
result = []
|
||||
for u in users:
|
||||
result.append({
|
||||
"id": getattr(u, "id", None),
|
||||
"name": getattr(u, "name", None),
|
||||
"email": getattr(u, "email", None),
|
||||
"display_name": getattr(u, "display_name", None),
|
||||
})
|
||||
return json.dumps({"count": len(result), "users": result}, indent=2)
|
||||
|
||||
async def _orgs(self) -> str:
|
||||
session = _get_session()
|
||||
iam = session.get_iam_client()
|
||||
orgs = iam.get_organizations()
|
||||
return json.dumps({"count": len(orgs), "organizations": orgs}, indent=2)
|
||||
|
||||
async def _roles(self) -> str:
|
||||
session = _get_session()
|
||||
iam = session.get_iam_client()
|
||||
roles = iam.get_roles()
|
||||
return json.dumps({"count": len(roles), "roles": roles}, indent=2)
|
||||
|
||||
# -- PaaS actions --------------------------------------------------------
|
||||
|
||||
async def _projects(self, org: str | None) -> str:
|
||||
if not org:
|
||||
return json.dumps({"error": "Required: org (organization ID)"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
projects = paas.get(f"/v1/org/{org}/project")
|
||||
return json.dumps({"org": org, "count": len(projects), "projects": projects}, indent=2)
|
||||
|
||||
async def _envs(self, org: str | None, project: str | None) -> str:
|
||||
if not org or not project:
|
||||
return json.dumps({"error": "Required: org and project"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
envs = paas.get(f"/v1/org/{org}/project/{project}/env")
|
||||
return json.dumps({"org": org, "project": project, "environments": envs}, indent=2)
|
||||
|
||||
async def _deployments(self, org: str | None, project: str | None, env: str | None) -> str:
|
||||
if not org or not project or not env:
|
||||
return json.dumps({"error": "Required: org, project, and environment"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
containers = paas.get(f"/v1/org/{org}/project/{project}/env/{env}/container")
|
||||
|
||||
result = []
|
||||
for c in containers if isinstance(containers, list) else []:
|
||||
result.append({
|
||||
"id": c.get("id"),
|
||||
"name": c.get("name"),
|
||||
"image": c.get("image"),
|
||||
"status": c.get("status"),
|
||||
"replicas": c.get("replicas"),
|
||||
})
|
||||
|
||||
return json.dumps({
|
||||
"org": org,
|
||||
"project": project,
|
||||
"environment": env,
|
||||
"count": len(result),
|
||||
"containers": result,
|
||||
}, indent=2)
|
||||
|
||||
async def _deploy_detail(
|
||||
self, org: str | None, project: str | None, env: str | None, container: str | None
|
||||
) -> str:
|
||||
if not org or not project or not env or not container:
|
||||
return json.dumps({"error": "Required: org, project, environment, and container"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
data = paas.get(f"/v1/org/{org}/project/{project}/env/{env}/container/{container}")
|
||||
return json.dumps(data, indent=2)
|
||||
|
||||
async def _logs(
|
||||
self, org: str | None, project: str | None, env: str | None, container: str | None
|
||||
) -> str:
|
||||
if not org or not project or not env or not container:
|
||||
return json.dumps({"error": "Required: org, project, environment, and container"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
logs = paas.get(f"/v1/org/{org}/project/{project}/env/{env}/container/{container}/logs")
|
||||
if isinstance(logs, dict):
|
||||
return json.dumps(logs, indent=2)
|
||||
return str(logs)
|
||||
|
||||
async def _redeploy(
|
||||
self, org: str | None, project: str | None, env: str | None, container: str | None
|
||||
) -> str:
|
||||
if not org or not project or not env or not container:
|
||||
return json.dumps({"error": "Required: org, project, environment, and container"})
|
||||
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
|
||||
# Fetch current config and PUT it back to trigger rolling restart
|
||||
current = paas.get(f"/v1/org/{org}/project/{project}/env/{env}/container/{container}")
|
||||
result = paas.put(
|
||||
f"/v1/org/{org}/project/{project}/env/{env}/container/{container}",
|
||||
json=current,
|
||||
)
|
||||
return json.dumps({"action": "redeployed", "container": container, "result": result}, indent=2)
|
||||
|
||||
# -- Cloud actions -------------------------------------------------------
|
||||
|
||||
async def _services(self) -> str:
|
||||
session = _get_session()
|
||||
paas = session.get_paas_client()
|
||||
info = paas.get("/v1/cluster/info")
|
||||
templates = paas.get("/v1/cluster/templates")
|
||||
return json.dumps({
|
||||
"cluster": info,
|
||||
"templates": templates,
|
||||
}, indent=2)
|
||||
|
||||
# -- Registration --------------------------------------------------------
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register platform tool with explicit parameters."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool(
|
||||
name="platform",
|
||||
description=DESCRIPTION,
|
||||
)
|
||||
async def platform(
|
||||
action: Annotated[
|
||||
str,
|
||||
Field(
|
||||
description=(
|
||||
"Action to perform. "
|
||||
"IAM: whoami, users, orgs, roles. "
|
||||
"PaaS: projects, env, deployments, deploy, logs, redeploy. "
|
||||
"Cloud: services."
|
||||
),
|
||||
),
|
||||
] = "whoami",
|
||||
org: Annotated[
|
||||
str | None,
|
||||
Field(description="Organization ID (for PaaS actions)"),
|
||||
] = None,
|
||||
project: Annotated[
|
||||
str | None,
|
||||
Field(description="Project ID (for PaaS actions)"),
|
||||
] = None,
|
||||
environment: Annotated[
|
||||
str | None,
|
||||
Field(description="Environment ID (for PaaS actions)"),
|
||||
] = None,
|
||||
container: Annotated[
|
||||
str | None,
|
||||
Field(description="Container/deployment ID (for deploy, logs, redeploy)"),
|
||||
] = None,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
return await tool_instance.call(
|
||||
ctx,
|
||||
action=action,
|
||||
org=org,
|
||||
project=project,
|
||||
environment=environment,
|
||||
container=container,
|
||||
)
|
||||
@@ -0,0 +1,33 @@
|
||||
[project]
|
||||
name = "hanzo-tools-platform"
|
||||
version = "0.1.0"
|
||||
description = "Hanzo platform MCP tool — PaaS deployments, IAM users/orgs, and cloud services"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
|
||||
keywords = ["hanzo", "mcp", "paas", "iam", "platform", "tools"]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"hanzo-tools-core>=0.1.0",
|
||||
"hanzo-tools-auth>=0.1.0",
|
||||
"hanzo-iam>=0.1.0",
|
||||
"httpx>=0.27.0",
|
||||
]
|
||||
|
||||
[project.entry-points."hanzo.tools"]
|
||||
platform = "hanzo_tools.platform:TOOLS"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["hanzo_tools"]
|
||||
Reference in New Issue
Block a user