Examples#
This section provides practical examples of how to use the Paramus Extension API through both access methods: JSON-RPC for external clients and direct Python for internal use.
JSON-RPC Examples (External Integration)#
Basic Test Commands#
Test Hello World#
import requests
import json
# Test the copilot system
payload = {
"jsonrpc": "2.0",
"method": "copilot.test.hello",
"id": 1
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
result = response.json()
print(result) # {"jsonrpc": "2.0", "result": {"status": "success", "message": "..."}, "id": 1}
Core System Test#
# Test the core system
payload = {
"jsonrpc": "2.0",
"method": "core.test.hello",
"id": 2
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
Terminal Communication#
Send Terminal Message#
# Send message with full control
payload = {
"jsonrpc": "2.0",
"method": "copilot.sendmessage.terminal",
"params": {
"content": "Processing complete!",
"wait_for_ws": True,
"retry_interval": 1.5
},
"id": 3
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
Quick Terminal Send#
# Quick send without WebSocket waiting
payload = {
"jsonrpc": "2.0",
"method": "copilot.quicksend.terminal",
"params": {
"content": "Status update"
},
"id": 4
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
UI Control Operations#
Send Notification#
# Show notification popup
payload = {
"jsonrpc": "2.0",
"method": "copilot.send.notification",
"params": {
"message": "Operation completed successfully!"
},
"id": 6
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
Update DOM Element#
# Update element properties
payload = {
"jsonrpc": "2.0",
"method": "copilot.update.element",
"params": {
"element_id": "status-indicator",
"property": "textContent",
"value": "System Online"
},
"id": 7
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
Get API Documentation#
# Get complete API documentation
payload = {
"jsonrpc": "2.0",
"method": "api.describe",
"id": 8
}
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
result = response.json()
# Access the documentation
if "result" in result and "documentation" in result["result"]:
docs = result["result"]["documentation"]
for class_name, class_info in docs.items():
print(f"Class: {class_name}")
if "jsonrpc_methods" in class_info:
for method, info in class_info["jsonrpc_methods"].items():
print(f" Method: {method} - {info['description']}")
Direct Python Examples (Internal Use)#
Using Command Registry Directly#
from paramus.extension.api_surface.commands import extension_commands
# Get the command registry
registry = extension_commands.get_command_registry()
# Execute commands directly
result = registry["copilot.test.hello"]()
print(result) # {"status": "success", "message": "Hello from UI commands"}
# With parameters
result = registry["copilot.sendmessage.terminal"]({
"content": "Direct command execution",
"wait_for_ws": False
})
print(result)
Using API Classes Directly#
from paramus.extension.api_surface.ui import MessageTerminal, UIBridge
from paramus.extension.api_surface.core import CoreTest
# MessageTerminal usage
terminal = MessageTerminal()
terminal.send_message("Hello from terminal!")
terminal.send_message_nowait("Quick message")
# Class method usage
MessageTerminal.quick_send("Static message")
# UIBridge usage
ui = UIBridge()
ui.trigger_new_button()
ui.trigger_sync_button()
ui.send_notification("Operation completed!")
ui.update_element_text("status", "Online")
# Class method usage
UIBridge.quick_trigger_new()
UIBridge.quick_notification("Status update")
# CoreTest usage
core = CoreTest()
core.print_hello_world()
# Class method usage
CoreTest.run()
LangChain Integration Examples#
Using Copilot Tools with LangChain#
from paramus.extension.host.agent.copilot_tools import (
get_all_copilot_tools,
get_copilot_tools_by_category
)
from langchain.agents import initialize_agent
from langchain.llms import OpenAI
# Get all available tools
tools = get_all_copilot_tools()
# Or get tools by category
categorized_tools = get_copilot_tools_by_category()
testing_tools = categorized_tools["testing"]
ui_tools = categorized_tools["ui_control"]
# Initialize LangChain agent with tools
llm = OpenAI(temperature=0)
agent = initialize_agent(
tools=tools,
llm=llm,
agent="zero-shot-react-description",
verbose=True
)
# Use the agent
result = agent.run("Test the copilot system and send a notification")
Individual Tool Usage#
from paramus.extension.host.agent.copilot_tools import (
CopilotTestHelloTool,
CopilotSendNotificationTool,
CoreTestHelloTool
)
# Create tool instances
hello_tool = CopilotTestHelloTool()
notification_tool = CopilotSendNotificationTool()
core_tool = CoreTestHelloTool()
# Use tools directly
result1 = hello_tool._run("")
print(f"Test result: {result1}")
result2 = notification_tool._run("Task completed successfully!")
print(f"Notification result: {result2}")
result3 = core_tool._run("")
print(f"Core test result: {result3}")
Error Handling Examples#
JSON-RPC Error Handling#
import requests
import json
def make_safe_jsonrpc_request(method, params=None):
"""Make a JSON-RPC request with proper error handling."""
payload = {
"jsonrpc": "2.0",
"method": method,
"id": 1
}
if params:
payload["params"] = params
try:
response = requests.post("http://localhost:8050/jsonrpc", json=payload)
result = response.json()
if "error" in result:
print(f"JSON-RPC Error: {result['error']['message']}")
return None
elif "result" in result:
return result["result"]
except requests.exceptions.ConnectionError:
print("Could not connect to JSON-RPC server")
return None
except json.JSONDecodeError:
print("Invalid JSON response")
return None
except Exception as e:
print(f"Unexpected error: {e}")
return None
# Usage
result = make_safe_jsonrpc_request("copilot.test.hello")
if result:
print(f"Success: {result}")
Direct Python Error Handling#
from paramus.extension.api_surface.commands import extension_commands
def execute_command_safely(method, params=None):
"""Execute a command with error handling."""
try:
registry = extension_commands.get_command_registry()
if method not in registry:
return {"status": "error", "message": f"Method {method} not found"}
command_func = registry[method]
result = command_func(params)
return result
except Exception as e:
return {"status": "error", "message": f"Execution error: {e}"}
# Usage
result = execute_command_safely("copilot.test.hello")
print(result)
result = execute_command_safely("invalid.method")
print(result) # {"status": "error", "message": "Method invalid.method not found"}
Custom Extension Example#
Creating a Custom API Class#
from typing import Dict, Any
from paramus.extension.api_surface.registry import ExtensionRegistry
@ExtensionRegistry.register
class MyCustomExtension:
"""Custom extension for specialized functionality."""
def __init__(self):
self.status = "initialized"
@classmethod
def describe(cls) -> Dict[str, Any]:
"""Return API documentation for this class."""
return {
"class": "MyCustomExtension",
"description": "Custom extension for specialized functionality",
"jsonrpc_methods": {
"custom.do_something": {
"description": "Execute custom functionality",
"parameters": {
"data": {"type": "str", "required": True, "description": "Input data"}
},
"example": "JSON-RPC call: custom.do_something"
}
},
"methods": {
"do_something": {
"description": "Execute custom functionality",
"parameters": {
"data": {"type": "str", "required": True, "description": "Input data"}
},
"example": "custom.do_something('input_data')"
}
}
}
def do_something(self, data: str) -> str:
"""Execute custom functionality."""
return f"Processed: {data}"
# Register the custom extension and add to command registry
# (This would be done in commands.py)
Integration with Command Registry#
# In commands.py, add handler for custom extension
def call_custom_do_something(self, params=None):
"""Call custom extension functionality."""
if not params or 'data' not in params:
return {"status": "error", "message": "data parameter required"}
custom_ext = MyCustomExtension()
result = custom_ext.do_something(params['data'])
return {"status": "success", "result": result}
# Add to get_command_registry method:
def get_command_registry(self):
return {
# ... existing commands ...
'custom.do_something': self.call_custom_do_something,
}
def describe(cls) -> Dict[str, Any]:
"""Return API documentation for this class."""
return {
"class": "MyCustomExtension",
"description": "Custom extension for specific functionality",
"methods": {
"do_work": {
"description": "Perform some work",
"parameters": {
"task": {"type": "str", "required": True, "description": "Task to perform"}
}
}
}
}
def do_work(self, task: str):
"""Perform some work based on the task."""
print(f"Performing task: {task}")
return f"Completed: {task}"
### Getting API Documentation
```python
from paramus.extension.api_surface.command import ExtensionRegistry
# Get documentation for all registered APIs
docs = ExtensionRegistry.describe_all()
print(docs)
# Get all registered classes
classes = ExtensionRegistry.get_all_classes()
for name, cls in classes.items():
print(f"Available API: {name}")
JSON-RPC Client Example#
import json
import socket
def send_jsonrpc_request(method, params=None):
"""Send a JSON-RPC request to the extension host."""
request = {
"jsonrpc": "2.0",
"method": method,
"params": params or {},
"id": 1
}
# This is a conceptual example - actual implementation would depend on
# your JSON-RPC transport mechanism
return json.dumps(request)
# Example requests
terminal_request = send_jsonrpc_request("terminal.send_message", {
"content": "Hello from JSON-RPC!"
})
ui_request = send_jsonrpc_request("ui.send_notification", {
"message": "Notification from JSON-RPC"
})
api_docs_request = send_jsonrpc_request("api.describe")
Integration Examples#
Extension with Error Handling#
from paramus.extension.api_surface.command import MessageTerminal, UIBridge
import logging
class SafeExtension:
def __init__(self):
self.terminal = MessageTerminal()
self.ui = UIBridge()
self.logger = logging.getLogger(__name__)
def safe_send_message(self, message: str) -> bool:
"""Safely send a message with error handling."""
try:
self.terminal.send_message(message)
self.ui.send_notification(f"Message sent: {message}")
return True
except Exception as e:
self.logger.error(f"Failed to send message: {e}")
self.ui.send_notification("Failed to send message")
return False
def batch_operations(self, messages: list):
"""Perform multiple operations with progress feedback."""
total = len(messages)
for i, message in enumerate(messages):
if self.safe_send_message(message):
progress = f"Progress: {i+1}/{total}"
self.ui.update_element_text("progress", progress)
else:
break
Asynchronous Extension#
import asyncio
from paramus.extension.api_surface.command import MessageTerminal
class AsyncExtension:
def __init__(self):
self.terminal = MessageTerminal()
async def send_delayed_messages(self, messages: list, delay: float = 1.0):
"""Send messages with delays between them."""
for message in messages:
await asyncio.sleep(delay)
self.terminal.send_message_nowait(f"Delayed: {message}")
async def run_async_task(self):
"""Example of running an asynchronous task."""
messages = ["Starting task", "Processing...", "Almost done", "Completed!"]
await self.send_delayed_messages(messages, 0.5)
# Usage
async def main():
ext = AsyncExtension()
await ext.run_async_task()
# Run with asyncio.run(main())