#!/usr/bin/env python3
"""
MCP Client for Claude Code Integration
Provides MCP tools that Claude Code can use directly
"""
import asyncio
import json
import aiohttp
from typing import Dict, Any, Optional
class MCPGeminiClient:
"""MCP client for Gemini collaboration"""
def __init__(self, server_url: str = "http://localhost:8080"):
self.server_url = server_url
self.conversation_id = None
async def start_collaboration(self, topic: str) -> Dict[str, Any]:
"""Start a new collaboration 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.conversation_id = data['conversation_id']
return data
async def consult_gemini(self, query: str, include_context: bool = True) -> Dict[str, Any]:
"""Consult Gemini with current conversation context"""
if not self.conversation_id:
# Auto-start if no conversation exists
await self.start_collaboration("General Discussion")
async with aiohttp.ClientSession() as session:
# Add Claude's message
await session.post(
f"{self.server_url}/add_message",
json={
"conversation_id": self.conversation_id,
"role": "claude",
"content": query
}
)
# Get Gemini's response
async with session.post(
f"{self.server_url}/consult_gemini",
json={
"conversation_id": self.conversation_id,
"query": query,
"include_context": include_context
}
) as response:
return await response.json()
async def get_conversation(self) -> Dict[str, Any]:
"""Get current conversation history"""
if not self.conversation_id:
return {"error": "No active conversation"}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.server_url}/conversation/{self.conversation_id}"
) as response:
return await response.json()
# Global client instance
mcp_client = MCPGeminiClient()
def mcp_start_collaboration(topic: str) -> str:
"""MCP tool: Start collaboration with Gemini"""
try:
result = asyncio.run(mcp_client.start_collaboration(topic))
if 'conversation_id' in result:
return f"Started collaboration on '{topic}' (ID: {result['conversation_id']})"
else:
return f"Failed to start collaboration: {result.get('error', 'Unknown error')}"
except Exception as e:
return f"Error starting collaboration: {str(e)}"
def mcp_consult_gemini(query: str) -> str:
"""MCP tool: Consult Gemini with a query"""
try:
result = asyncio.run(mcp_client.consult_gemini(query))
if 'gemini_response' in result:
gemini_data = result['gemini_response']
if gemini_data.get('error'):
return f"Gemini error: {gemini_data.get('message', 'Unknown error')}"
else:
return gemini_data.get('text', 'No response from Gemini')
else:
return f"Failed to consult Gemini: {result.get('error', 'Unknown error')}"
except Exception as e:
return f"Error consulting Gemini: {str(e)}"
def mcp_get_conversation() -> str:
"""MCP tool: Get conversation history"""
try:
result = asyncio.run(mcp_client.get_conversation())
if 'messages' in result:
messages = result['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 f"Failed to get conversation: {result.get('error', 'No conversation found')}"
except Exception as e:
return f"Error getting conversation: {str(e)}"
# Test the tools
if __name__ == "__main__":
print("Testing MCP tools...")
print("1. Starting collaboration:", mcp_start_collaboration("Test topic"))
print("2. Consulting Gemini:", mcp_consult_gemini("What's your favorite color?"))
print("3. Getting conversation:", mcp_get_conversation())
