#!/usr/bin/env python3
"""
Claude-Gemini Collaboration Tools
Functions for Claude to use during conversations with Gemini
"""
import asyncio
import json
import aiohttp
from typing import Dict, Any, Optional
import uuid
class GeminiCollaborator:
"""Tool for Claude to collaborate with Gemini"""
def __init__(self, server_url: str = "http://localhost:8080"):
self.server_url = server_url
self.current_conversation_id = None
async def start_collaboration(self, topic: str) -> str:
"""Start a new collaborative conversation with Gemini"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.server_url}/start_conversation",
json={"topic": topic}
) as response:
data = await response.json()
if response.status == 200:
self.current_conversation_id = data['conversation_id']
return f"Started collaboration on '{topic}' (ID: {self.current_conversation_id})"
else:
return f"Failed to start collaboration: {data.get('error', 'Unknown error')}"
async def discuss_with_gemini(self, query: str, include_context: bool = True) -> str:
"""Discuss a topic with Gemini, maintaining conversation context"""
if not self.current_conversation_id:
return "No active conversation. Please start a collaboration first."
async with aiohttp.ClientSession() as session:
# Add Claude's message to conversation
await session.post(
f"{self.server_url}/add_message",
json={
"conversation_id": self.current_conversation_id,
"role": "claude",
"content": query
}
)
# Get Gemini's response
async with session.post(
f"{self.server_url}/consult_gemini",
json={
"conversation_id": self.current_conversation_id,
"query": query,
"include_context": include_context
}
) as response:
data = await response.json()
if response.status == 200:
gemini_response = data.get('gemini_response', {})
if gemini_response.get('error'):
return f"Gemini error: {gemini_response.get('message', 'Unknown error')}"
else:
return gemini_response.get('text', 'No response from Gemini')
else:
return f"Failed to consult Gemini: {data.get('error', 'Unknown error')}"
async def get_conversation_history(self) -> str:
"""Get the full conversation history"""
if not self.current_conversation_id:
return "No active conversation."
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.server_url}/conversation/{self.current_conversation_id}"
) as response:
if response.status == 200:
data = await response.json()
messages = data.get('messages', [])
history = []
for msg in messages:
role = msg['role'].upper()
content = msg['content']
history.append(f"{role}: {content}")
return "\n\n".join(history)
else:
return "Failed to get conversation history"
async def iterative_planning(self, initial_plan: str, iterations: int = 3) -> str:
"""Iteratively refine a plan with Gemini"""
if not self.current_conversation_id:
await self.start_collaboration("Iterative Planning")
current_plan = initial_plan
results = [f"INITIAL PLAN (Claude):\n{current_plan}"]
for i in range(iterations):
# Ask Gemini to review and improve the plan
gemini_query = f"Please review this plan and suggest improvements:\n\n{current_plan}\n\nWhat would you add, remove, or modify?"
gemini_response = await self.discuss_with_gemini(gemini_query)
results.append(f"GEMINI FEEDBACK (Round {i+1}):\n{gemini_response}")
# Claude synthesizes the feedback
claude_synthesis = f"Based on Gemini's feedback, here's the refined plan:\n\n{current_plan}\n\nIncorporating suggestions: {gemini_response}"
# Update current plan (in real implementation, Claude would actually refine this)
current_plan = f"REFINED PLAN (Round {i+1}):\n{claude_synthesis}"
results.append(current_plan)
return "\n\n" + "="*50 + "\n\n".join(results)
# Global instance for easy access
gemini_collaborator = GeminiCollaborator()
# Convenience functions that Claude can call directly
async def start_collaboration_with_gemini(topic: str) -> str:
"""Start collaborating with Gemini on a topic"""
return await gemini_collaborator.start_collaboration(topic)
async def ask_gemini(query: str) -> str:
"""Ask Gemini a question in the current conversation"""
return await gemini_collaborator.discuss_with_gemini(query)
async def brainstorm_with_gemini(topic: str) -> str:
"""Brainstorm ideas with Gemini"""
await gemini_collaborator.start_collaboration(f"Brainstorming: {topic}")
return await gemini_collaborator.discuss_with_gemini(f"Let's brainstorm ideas for: {topic}. What are your initial thoughts?")
async def get_gemini_review(content: str) -> str:
"""Get Gemini's review of content"""
return await gemini_collaborator.discuss_with_gemini(f"Please review this and provide feedback:\n\n{content}")
async def iterative_plan_with_gemini(initial_plan: str, rounds: int = 3) -> str:
"""Iteratively refine a plan with Gemini"""
return await gemini_collaborator.iterative_planning(initial_plan, rounds)
# Test function
async def test_collaboration():
"""Test the collaboration system"""
print("Testing collaboration system...")
# Start collaboration
result = await start_collaboration_with_gemini("Testing the collaboration system")
print(f"Start: {result}")
# Ask a question
response = await ask_gemini("What do you think about this collaboration system?")
print(f"Gemini: {response}")
# Get history
history = await gemini_collaborator.get_conversation_history()
print(f"History:\n{history}")
if __name__ == "__main__":
asyncio.run(test_collaboration())
