Python Client Example
A minimal Python client library for interacting with a Skills Runtime. This example demonstrates the standard JSON-RPC 2.0 protocol and typical usage patterns.
SkillsRuntimeClient
Here's a complete client implementation:
import requests
import json
class SkillsRuntimeClient:
def __init__(self, endpoint="http://localhost:8080/rpc"):
self.endpoint = endpoint
self.request_id = 0
def _call(self, method, params=None):
"""Make a JSON-RPC 2.0 call to the runtime."""
self.request_id += 1
payload = {
"jsonrpc": "2.0",
"id": str(self.request_id),
"method": method,
"params": params or {}
}
response = requests.post(self.endpoint, json=payload)
response.raise_for_status()
result = response.json()
if "error" in result:
raise RuntimeError(f"RPC error: {result['error']}")
return result.get("result")
def list_skills(self, namespace=None, detail="names", limit=50):
"""List available skills."""
params = {"detail": detail, "limit": limit}
if namespace:
params["namespace"] = namespace
return self._call("list_skills", params)
def describe_skill(self, name, version=None, detail="summary"):
"""Get skill manifest and documentation."""
params = {"name": name, "detail": detail}
if version:
params["version"] = version
return self._call("describe_skill", params)
def read_skill_file(self, name, path, version=None):
"""Read a file from a skill's directory."""
params = {"name": name, "path": path}
if version:
params["version"] = version
return self._call("read_skill_file", params)
def execute_skill(self, name, args=None, version=None, input_blobs=None, timeout_ms=None):
"""Execute a skill."""
params = {"name": name}
if version:
params["version"] = version
if args:
params["args"] = args
if input_blobs:
params["input_blobs"] = input_blobs
if timeout_ms:
params["timeout_ms"] = timeout_ms
return self._call("execute_skill", params)
def run_code(self, code, language="python", entrypoint="main",
args=None, mount_skills=None, input_blobs=None, timeout_ms=None):
"""Execute custom code with mounted skills."""
params = {
"language": language,
"code": code,
"entrypoint": entrypoint,
"args": args or {}
}
if mount_skills:
params["mount_skills"] = mount_skills
if input_blobs:
params["input_blobs"] = input_blobs
if timeout_ms:
params["limits"] = {"timeout_ms": timeout_ms}
return self._call("run_code", params)
def create_blob(self, content, kind="text/plain"):
"""Store content as a blob."""
return self._call("create_blob", {"content": content, "kind": kind})
def read_blob(self, blob_id, mode="sample_head", max_bytes=2000):
"""Read a blob preview."""
return self._call("read_blob", {
"blob_id": blob_id,
"mode": mode,
"max_bytes": max_bytes
})
def load_protocol_guide(self):
"""Load the Skills Protocol Guide."""
return self._call("load_skills_protocol_guide", {})
Basic methods
Initialize the client
client = SkillsRuntimeClient(endpoint="http://localhost:8080/rpc")
Load the protocol guide
guide = client.load_protocol_guide()
print(guide["content"])
List available skills
skills = client.list_skills(namespace="salesforce", detail="summary")
for skill in skills["skills"]:
print(f"- {skill['name']}: {skill['description']}")
Describe a skill
info = client.describe_skill("salesforce.leads.sync")
print(f"Manifest: {info['skill']['manifest']}")
Execute a skill
result = client.execute_skill(
name="salesforce.leads.sync",
args={"sheet_blob": "blob:abc123", "env": "prod"},
input_blobs=["blob:abc123"]
)
if result["status"] == "completed":
print(f"Success: {result['summary']}")
else:
print(f"Failed: {result['error']['message']}")
Complete example
Here's a complete workflow:
if __name__ == "__main__":
client = SkillsRuntimeClient()
# 1. Load the protocol guide (first time)
print("=== Loading Protocol Guide ===")
guide = client.load_protocol_guide()
print(guide["content"][:300], "...\n")
# 2. List skills in a namespace
print("=== Listing Salesforce Skills ===")
skills = client.list_skills(namespace="salesforce", detail="summary")
print(json.dumps(skills, indent=2), "\n")
# 3. Get skill details
print("=== Skill Details ===")
skill_info = client.describe_skill("salesforce.leads.sync")
print(json.dumps(skill_info, indent=2), "\n")
# 4. Read full documentation
print("=== Full Documentation ===")
docs = client.read_skill_file("salesforce.leads.sync", "SKILL.md")
print(docs["content"], "\n")
# 5. Create a blob with large data
print("=== Creating Blob ===")
csv_data = "name,email,company\nJohn Doe,john@example.com,Acme Corp"
blob_result = client.create_blob(csv_data, kind="text/csv")
blob_id = blob_result["blob_id"]
print(f"Created blob: {blob_id}\n")
# 6. Execute a skill
print("=== Executing Skill ===")
result = client.execute_skill(
name="salesforce.leads.sync",
args={"sheet_blob": blob_id, "env": "prod"},
input_blobs=[blob_id]
)
print(json.dumps(result, indent=2), "\n")
# 7. Run custom code with multiple skills
print("=== Running Custom Code ===")
code = """
from skills.salesforce.leads import sync_leads
from runtime import blobs, log
def main(args):
log.info("Starting custom sync workflow...")
sheet_data = blobs.read_text(args['sheet_blob'])
result = sync_leads(sheet_data, env=args['env'])
return {
"status": "completed",
"synced_count": result['count']
}
"""
code_result = client.run_code(
code=code,
args={"sheet_blob": blob_id, "env": "prod"},
mount_skills=["salesforce.leads.sync"],
input_blobs=[blob_id]
)
print(json.dumps(code_result, indent=2))
Error handling
The client raises RuntimeError for RPC-level errors. Handle them appropriately:
try:
result = client.execute_skill(name="nonexistent.skill", args={})
except RuntimeError as e:
print(f"Error: {e}")
# For run failures (different from RPC errors):
result = client.execute_skill(name="salesforce.leads.sync", args={})
if result["status"] == "failed":
print(f"Skill execution failed: {result['error']['message']}")