pipe_client.py•4.16 kB
#!/usr/bin/env python3
"""
Lightweight wrapper for interacting with Katamari MCP named pipes.
Provides simple CLI interface for pipe communication with context persistence.
"""
import asyncio
import json
import sys
import argparse
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from katamari_mcp.transport.named_pipe import send_pipe_request
async def main():
"""Main CLI interface."""
parser = argparse.ArgumentParser(description="Katamari MCP Pipe Client")
parser.add_argument('command', choices=[
'list', 'call', 'status', 'add', 'check', 'cancel'
], help='Command to execute')
parser.add_argument('--tool', help='Tool name for call command')
parser.add_argument('--args', help='Arguments for call command (JSON string)')
parser.add_argument('--task', help='Task ID for check/cancel commands')
parser.add_argument('--pipe', default='katamari_mcp', help='Named pipe name')
args = parser.parse_args()
try:
if args.command == 'list':
result = await send_pipe_request('list_capabilities', {}, args.pipe)
print("Available capabilities:")
for cap in result.get('result', {}).get('capabilities', []):
print(f" - {cap.get('name')}: {cap.get('description')}")
elif args.command == 'call':
if not args.tool:
print("Error: --tool required for call command")
return 1
tool_args = {}
if args.args:
try:
tool_args = json.loads(args.args)
except json.JSONDecodeError:
print("Error: Invalid JSON in --args")
return 1
result = await send_pipe_request('call_tool', {
'tool_name': args.tool,
'arguments': tool_args
}, args.pipe)
if 'error' in result:
print(f"Error: {result['error']}")
return 1
else:
print(f"Result: {result.get('result', {})}")
elif args.command == 'status':
result = await send_pipe_request('get_status', {}, args.pipe)
if 'error' in result:
print(f"Error: {result['error']}")
return 1
else:
status = result.get('result', {})
print(f"Server Status: {status.get('status')}")
print(f"Pipe Path: {status.get('pipe_path')}")
print(f"Context Size: {status.get('context_size')} items")
elif args.command == 'add':
# TaskMaster stretch goal - placeholder
print("TaskMaster feature not yet implemented")
print("This will allow adding background tasks like:")
print(" katamari add 'train_model' --args '{\"dataset\": \"data.csv\"}'")
return 1
elif args.command == 'check':
# TaskMaster stretch goal - placeholder
if not args.task:
print("Error: --task required for check command")
return 1
print("TaskMaster feature not yet implemented")
print("This will allow checking task status like:")
print(f" katamari check '{args.task}'")
return 1
elif args.command == 'cancel':
# TaskMaster stretch goal - placeholder
if not args.task:
print("Error: --task required for cancel command")
return 1
print("TaskMaster feature not yet implemented")
print("This will allow canceling tasks like:")
print(f" katamari cancel '{args.task}'")
return 1
return 0
except Exception as e:
print(f"Error: {e}")
return 1
if __name__ == "__main__":
try:
exit_code = asyncio.run(main())
sys.exit(exit_code)
except KeyboardInterrupt:
print("\nInterrupted")
sys.exit(1)