Writing Skills
writing-skills
效率与协作技能,来自 obra/superpowers,查看原文获取完整用途和操作说明。
multi-platform-agent-deployment【部署与运维】Deploys and orchestrates an identical agent across multiple execution environments (CLI, web UI, API, chat platforms) with shared knowledge base and behavioral consistency.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
Deploys and orchestrates an identical agent across multiple execution environments (CLI, web UI, API, chat platform) with a shared knowledge base and consistent behavior. This skill makes the model design environment-specific adapter layers, configuration synchronization, centralized capability registries, and deployment topology patterns for heterogeneous agent deployments.
Use this skill when:
Avoid this skill for:
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class Platform(str, Enum):
"""Supported interaction platforms for the agent."""
CLI = "cli"
WEB_UI = "web_ui"
REST_API = "rest_api"
CHAT_SLACK = "chat_slack"
CHAT_DISCORD = "chat_discord"
@dataclass
class AgentRequest:
"""Platform-independent input to the agent."""
user_id: str
message: str
context: dict[str, Any] = field(default_factory=dict)
session_id: str | None = None
@dataclass
class AgentResponse:
"""Platform-independent output from the agent."""
text: str
metadata: dict[str, Any] = field(default_factory=dict)
error_code: str | None = None
suggestion: str | None = None
@dataclass
class PlatformError:
"""Structured error response for platform consumption."""
code: str # e.g., "RATE_LIMITED", "INVALID_INPUT", "INTERNAL_ERROR"
message: str
suggestion: str | None = None
class AgentAdapter(ABC):
"""Protocol that every platform adapter must implement.
Subclasses translate between platform-specific I/O formats and the
platform-independent AgentRequest / AgentResponse types so that the
core agent service never sees HTTP, TTY, or chat-protocol details.
"""
def __init__(self, core_agent: Any) -> None:
"""Initialize with a reference to the core agent service.
Args:
core_agent: The platform-independent agent implementation.
"""
self._core_agent = core_agent
@abstractmethod
def parse_input(self, raw_input: Any) -> AgentRequest:
"""Convert platform-specific input to structured request.
Args:
raw_input: Raw data from the platform (string, dict, HTTP body, etc.).
Returns:
A fully-formed AgentRequest instance.
"""
...
@abstractmethod
def format_output(self, response: AgentResponse) -> Any:
"""Convert structured agent output back to platform format.
Args:
response: The platform-independent AgentResponse from the core agent.
Returns:
Data formatted for the target platform (CLI string, JSON body, etc.).
"""
...
@abstractmethod
def get_platform_error_response(self, error: PlatformError) -> Any:
"""Convert a structured error to platform-specific format.
Args:
error: The structured PlatformError with code and message.
Returns:
An error response appropriate for the platform.
"""
...
async def handle_request(self, raw_input: Any) -> Any:
"""Full request lifecycle: parse → execute → format.
Orchestrates input parsing, core agent execution, and output formatting
while catching errors at every stage and converting them to structured
PlatformError responses.
Args:
raw_input: Raw data from the platform.
Returns:
Platform-formatted response or error output.
"""
try:
request = self.parse_input(raw_input)
response = await self._core_agent.execute(request)
return self.format_output(response)
except ValueError as e:
error = PlatformError(code="INVALID_INPUT", message=str(e))
return self.get_platform_error_response(error)
except Exception as e:
error = PlatformError(
code="INTERNAL_ERROR",
message="Agent processing failed. Please retry.",
suggestion="If the issue persists, contact support.",
)
return self.get_platform_error_response(error)
import sys
import json
from dataclasses import dataclass
from multiplatform.adapters.base import AgentAdapter, AgentResponse, PlatformError
class CLIAdapter(AgentAdapter):
"""Adapter for command-line interface interaction.
Accepts raw text lines from stdin or JSON objects piped through stdin.
Outputs formatted strings suitable for terminal display.
"""
def parse_input(self, raw_input: str) -> AgentRequest:
"""Parse CLI input — either raw text or JSON from stdin.
If the input starts with '{', attempts a JSON decode and extracts
user_id, message, context, and session_id fields. Falls back to
treating the entire string as a plain-text message otherwise.
Args:
raw_input: A single line of CLI input.
Returns:
An AgentRequest derived from the parsed input.
Raises:
ValueError: If JSON is malformed and fallback extraction fails.
"""
if raw_input.startswith("{"):
try:
data = json.loads(raw_input)
return AgentRequest(
user_id=data.get("user_id", "cli"),
message=data["message"],
context=data.get("context", {}),
)
except (json.JSONDecodeError, KeyError):
# Fall back to raw text
pass
return AgentRequest(
user_id="cli",
message=raw_input.strip(),
)
def format_output(self, response: AgentResponse) -> str:
"""Format output for CLI display.
Appends metadata as indented key-value pairs when present and
prefixes error responses with emoji markers for visual clarity.
Args:
response: The platform-independent agent response.
Returns:
A human-readable string suitable for terminal printing.
"""
if response.error_code:
return f"Error [{response.error_code}]: {response.message}\nSuggestion: {response.suggestion}"
# Pretty-print with metadata
parts = [response.text]
if response.metadata:
parts.append(f"\n--- Metadata ---")
for key, value in response.metadata.items():
parts.append(f" {key}: {value}")
return "\n".join(parts)
def get_platform_error_response(self, error: PlatformError) -> str:
"""Format a structured error for CLI display with emoji markers.
Args:
error: The structured error with code and message.
Returns:
A human-readable error string with suggestion fallback.
"""
return f"❌ [{error.code}] {error.message}\n💡 {error.suggestion or 'Try again.'}"
# Usage — run the CLI adapter as a stdin loop
if __name__ == "__main__": # pragma: no cover
cli = CLIAdapter(my_core_agent)
for line in sys.stdin:
result = cli.handle_request(line)
print(result)
from typing import Any
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="Agent Service API")
class APIRequest(BaseModel):
"""Schema for incoming agent requests via REST."""
user_id: str
message: str
context: dict[str, Any] = {}
session_id: str | None = None
class APIResponse(BaseModel):
"""Schema for outgoing agent responses via REST."""
text: str
metadata: dict[str, Any] = {}
error_code: str | None = None
suggestion: str | None = None
# Set at startup with core agent reference
api_adapter: Any = None # type: ignore[assignment]
@app.post("/api/v1/agent", response_model=APIResponse)
async def agent_endpoint(request: APIRequest) -> APIResponse:
"""REST endpoint for agent interaction.
Forwards the deserialized request to the core agent service and returns
the structured response with metadata.
Args:
request: Deserialized APIRequest from the client.
Returns:
An APIResponse matching the response model schema.
Raises:
HTTPException: On internal agent errors (500).
"""
try:
req = AgentRequest(
user_id=request.user_id,
message=request.message,
context=request.context,
session_id=request.session_id,
)
response = await api_adapter._core_agent.execute(req)
return APIResponse(text=response.text, metadata=response.metadata)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/health")
async def health_check() -> dict[str, str]:
"""Health check endpoint for load balancers and orchestrators.
Returns:
A dictionary with status and platform identifier.
"""
return {"status": "healthy", "platform": "rest_api"}
import os
import time
from typing import Any, Callable
class ConfigSyncService:
"""Ensures all platform adapters share identical configuration.
Loads configuration from the configured source (environment variables by
default), detects changes on a periodic basis, and notifies registered
listeners so that every platform instance stays in sync.
Attributes:
_config_source: The name of the configuration source to use.
_listeners: Callbacks invoked when configuration changes are detected.
_current_config: The most recently loaded configuration snapshot.
"""
def __init__(self, config_source: str = "environment") -> None:
"""Initialize the synchronization service.
Args:
config_source: Source identifier (e.g., 'environment', 'vault', 's3').
"""
self._config_source = config_source
self._listeners: list[Callable[[dict[str, Any]], None]] = []
self._current_config: dict[str, Any] = {}
def load_config(self) -> dict[str, Any]:
"""Load configuration from the configured source.
Reads standard environment variables with sensible defaults for common
agent settings such as model selection, token limits, temperature,
knowledge base URL, and rate limits.
Returns:
A dictionary of configuration key-value pairs.
"""
return {
"agent_model": os.getenv("AGENT_MODEL", "gpt-4o"),
"max_tokens": int(os.getenv("MAX_TOKENS", "4096")),
"temperature": float(os.getenv("TEMPERATURE", "0.7")),
"knowledge_base_url": os.getenv("KB_URL", "http://localhost:6333"),
"rate_limit_per_minute": int(os.getenv("RATE_LIMIT", "60")),
}
def register_listener(self, callback: Callable[[dict[str, Any]], None]) -> None:
"""Register a function to be called when config changes are detected.
The callback receives the new configuration dictionary and is invoked
for every distinct change while the sync loop is running.
Args:
callback: A callable accepting a configuration dict.
"""
self._listeners.append(callback)
def start_sync(self, interval_seconds: int = 60) -> None:
"""Periodically check for config changes and notify listeners.
Runs an infinite loop that polls the configuration source at the given
interval and triggers listeners when a difference is detected.
Args:
interval_seconds: Seconds between each configuration poll.
"""
while True:
new_config = self.load_config()
if new_config != self._current_config:
self._current_config = new_config
for listener in self._listeners:
listener(new_config)
time.sleep(interval_seconds)
# Usage — register all adapters as config listeners
if __name__ == "__main__": # pragma: no cover
sync = ConfigSyncService("environment")
sync.register_listener(
lambda config: print(f"Config updated: {config['agent_model']}")
)
sync.start_sync(interval_seconds=30)
code-philosophy (5 Laws of Elegant Defense): early exit on configuration mismatch, fail fast when a platform adapter deviates from the protocolWhen this skill is active, deliver:
| Skill | Purpose |
|---|---|
cli-agent-workflows | The CLI adapter is an implementation of this skill's patterns |
gui-agent-interaction | Web UI deployment uses GUI interaction concepts for the frontend |
mcp-integration | MCP servers can serve as the centralized knowledge base |
📖 skill(local cache): cli-agent-workflows, gui-agent-interaction, mcp-integration