Create kilo_client.py

SKILLS/kilo_agent/references/kilo_client.py

@@ -0,0 +1,124 @@
+import os  
+import requests  
+import json  
+import subprocess  
+import time  
+import re  
+from typing import Optional  
+  
+class KiloLocalClient:  
+    def __init__(self, base_url: str, password: Optional[str] = None, directory: Optional[str] = None):  
+        self.base_url = base_url.rstrip('/')  
+        self.session = requests.Session()  
+          
+        # Set up basic auth if password provided  
+        if password:  
+            self.session.auth = ('kilo', password)  
+          
+        # Set directory header if provided  
+        if directory:  
+            self.session.headers.update({'x-kilo-directory': directory})  
+      
+    def create_session(self, title: Optional[str] = None) -> dict:  
+        """Create a new session"""  
+        data = {}  
+        if title:  
+            data['title'] = title  
+          
+        response = self.session.post(f"{self.base_url}/session", json=data)  
+        response.raise_for_status()  
+        return response.json()  
+      
+    def send_message(self, session_id: str, message: str) -> dict:  
+        """Send a message to a session"""  
+        data = {  
+            'parts': [{'type': 'text', 'text': message}]  
+        }  
+        response = self.session.post(  
+            f"{self.base_url}/session/{session_id}/prompt",  
+            json=data  
+        )  
+        response.raise_for_status()  
+        return response.json()  
+      
+    def get_events(self, session_id: str):  
+        """Get event stream for a session"""  
+        response = self.session.get(f"{self.base_url}/global/event", stream=True)  
+        response.raise_for_status()  
+          
+        for line in response.iter_lines():  
+            if line:  
+                line = line.decode('utf-8')  
+                if line.startswith('data: '):  
+                    yield json.loads(line[6:])  
+  
+def start_local_server() -> tuple[str, str]:  
+    """Start a local kilo server and return (url, password)"""  
+    # Generate random password  
+    password = os.urandom(16).hex()  
+      
+    # Start server in background  
+    cmd = [  
+        'bun', 'dev', 'serve', '--port', '0'  
+    ]  
+      
+    env = os.environ.copy()  
+    env['KILO_SERVER_PASSWORD'] = password  
+      
+    with open('/tmp/kilo-serve.log', 'w') as f:  
+        process = subprocess.Popen(  
+            cmd, env=env, stdout=f, stderr=subprocess.STDOUT  
+        )  
+      
+    # Wait for server to start  
+    for _ in range(50):  # Wait up to 5 seconds  
+        time.sleep(0.1)  
+        try:  
+            with open('/tmp/kilo-serve.log', 'r') as f:  
+                content = f.read()  
+                match = re.search(r'listening on http://[^:]+:(\d+)', content)  
+                if match:  
+                    port = match.group(1)  
+                    return f"http://127.0.0.1:{port}", password  
+        except FileNotFoundError:  
+            pass  
+      
+    raise RuntimeError("Server failed to start")  
+  
+# Usage example  
+def main():  
+    # Start server  
+    server_url, password = start_local_server()  
+    print(f"Server started at {server_url}")  
+      
+    # Create client  
+    client = KiloLocalClient(  
+        base_url=server_url,  
+        password=password,  
+        directory=os.getcwd()  
+    )  
+      
+    # Create session  
+    session = client.create_session(title="Python Test Session")  
+    session_id = session['id']  
+    print(f"Created session: {session_id}")  
+      
+    # Send message  
+    response = client.send_message(  
+        session_id=session_id,  
+        message="List the files in the current directory"  
+    )  
+    print("Message sent, waiting for response...")  
+      
+    # Listen for events  
+    for event in client.get_events(session_id):  
+        if event.get('payload', {}).get('type') == 'session.message':  
+            content = event['payload'].get('content', [])  
+            for part in content:  
+                if part.get('type') == 'text':  
+                    print(part.get('text', ''))  
+        elif event.get('payload', {}).get('type') == 'session.idle':  
+            break  
+  
+if __name__ == "__main__":  
+    main()