Merge pull request #1 from gperdrizet/dev

.gitignore

@@ -1,3 +1,4 @@
+__pycache__
 .vscode
 .venv
 logs

client/anthropic_bridge.py

@@ -0,0 +1,303 @@
+'''Classes to connect to Anthropic inference endpoint'''
+
+import abc
+from typing import Dict, List, Any, Optional
+import anthropic
+from client.mcp_client import MCPClientWrapper, ToolDef, ToolParameter, ToolInvocationResult
+
+DEFAULT_ANTHROPIC_MODEL = 'claude-3-haiku-20240307'
+
+# Type mapping from Python/MCP types to JSON Schema types
+TYPE_MAPPING = {
+    'int': 'integer',
+    'bool': 'boolean',
+    'str': 'string',
+    'float': 'number',
+    'list': 'array',
+    'dict': 'object',
+    'boolean': 'boolean',
+    'string': 'string',
+    'integer': 'integer',
+    'number': 'number',
+    'array': 'array',
+    'object': 'object'
+}
+
+
+class LLMBridge(abc.ABC):
+    '''Abstract base class for LLM bridge implementations.'''
+
+    def __init__(self, mcp_client: MCPClientWrapper):
+        '''Initialize the LLM bridge with an MCPClient instance.
+
+        Args:
+            mcp_client: An initialized MCPClient instance
+        '''
+        self.mcp_client = mcp_client
+        self.tools = None
+
+
+    async def fetch_tools(self) -> List[ToolDef]:
+        '''Fetch available tools from the MCP endpoint.
+
+        
+        Returns:
+            List of ToolDef objects
+        '''
+        self.tools = await self.mcp_client.list_tools()
+        return self.tools
+
+
+    @abc.abstractmethod
+    async def format_tools(self, tools: List[ToolDef]) -> Any:
+        '''Format tools for the specific LLM provider.
+
+        Args:
+            tools: List of ToolDef objects
+            
+        Returns:
+            Formatted tools in the LLM-specific format
+        '''
+        pass
+
+
+    @abc.abstractmethod
+    async def submit_query(self, query: str, formatted_tools: Any) -> Dict[str, Any]:
+        '''Submit a query to the LLM with the formatted tools.
+
+        Args:
+            query: User query string
+            formatted_tools: Tools in the LLM-specific format
+            
+        Returns:
+            LLM response
+        '''
+        pass
+
+
+    @abc.abstractmethod
+    async def parse_tool_call(self, llm_response: Any) -> Optional[Dict[str, Any]]:
+        '''Parse the LLM response to extract tool calls.
+
+        Args:
+            llm_response: Response from the LLM
+            
+        Returns:
+            Dictionary with tool name and parameters, or None if no tool call
+        '''
+        pass
+
+
+    async def execute_tool(self, tool_name: str, kwargs: Dict[str, Any]) -> ToolInvocationResult:
+        '''Execute a tool with the given parameters.
+
+        Args:
+            tool_name: Name of the tool to invoke
+            kwargs: Dictionary of parameters to pass to the tool
+            
+        Returns:
+            ToolInvocationResult containing the tool's response
+        '''
+        return await self.mcp_client.invoke_tool(tool_name, kwargs)
+
+
+    async def process_query(self, query: str) -> Dict[str, Any]:
+        '''Process a user query through the LLM and execute any tool calls.
+
+        This method handles the full flow:
+        1. Fetch tools if not already fetched
+        2. Format tools for the LLM
+        3. Submit query to LLM
+        4. Parse tool calls from LLM response
+        5. Execute tool if needed
+        
+        Args:
+            query: User query string
+            
+        Returns:
+            Dictionary containing the LLM response, tool call, and tool result
+        '''
+        # 1. Fetch tools if not already fetched
+        if self.tools is None:
+            await self.fetch_tools()
+
+        # 2. Format tools for the LLM
+        formatted_tools = await self.format_tools(self.tools)
+
+        # 3. Submit query to LLM
+        llm_response = await self.submit_query(query, formatted_tools)
+
+        # 4. Parse tool calls from LLM response
+        tool_call = await self.parse_tool_call(llm_response)
+
+        result = {
+            'llm_response': llm_response,
+            'tool_call': tool_call,
+            'tool_result': None
+        }
+
+        # 5. Execute tool if needed
+        if tool_call:
+            tool_name = tool_call.get('name')
+            kwargs = tool_call.get('parameters', {})
+            tool_result = await self.execute_tool(tool_name, kwargs)
+            result['tool_result'] = tool_result
+
+        return result
+
+
+class AnthropicBridge(LLMBridge):
+    '''Anthropic-specific implementation of the LLM Bridge.'''
+
+    def __init__(self, mcp_client, api_key, model=DEFAULT_ANTHROPIC_MODEL): # Use imported default
+        '''Initialize Anthropic bridge with API key and model.
+        
+        Args:
+            mcp_client: An initialized MCPClient instance
+            api_key: Anthropic API key
+            model: Anthropic model to use (default: from models.py)
+        '''
+        super().__init__(mcp_client)
+        self.llm_client = anthropic.Anthropic(api_key=api_key)
+        self.model = model
+
+
+    async def format_tools(self, tools: List[ToolDef]) -> List[Dict[str, Any]]:
+        '''Format tools for Anthropic.
+        
+        Args:
+            tools: List of ToolDef objects
+            
+        Returns:
+            List of tools in Anthropic format
+        '''
+        return to_anthropic_format(tools)
+
+
+    async def submit_query(
+            self,
+            query: str,
+            formatted_tools: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
+        '''Submit a query to Anthropic with the formatted tools.
+        
+        Args:
+            query: User query string
+            formatted_tools: Tools in Anthropic format
+            
+        Returns:
+            Anthropic API response
+        '''
+        response = self.llm_client.messages.create(
+            model=self.model,
+            max_tokens=4096,
+            system='You are a helpful tool-using assistant.',
+            # messages=[
+            #     {'role': 'user', 'content': query}
+            # ],
+            messages=query,
+            tools=formatted_tools
+        )
+
+        return response
+
+
+    async def parse_tool_call(self, llm_response: Any) -> Optional[Dict[str, Any]]:
+        '''Parse the Anthropic response to extract tool calls.
+        
+        Args:
+            llm_response: Response from Anthropic
+            
+        Returns:
+            Dictionary with tool name and parameters, or None if no tool call
+        '''
+        for content in llm_response.content:
+            if content.type == 'tool_use':
+                return {
+                    'name': content.name,
+                    'parameters': content.input
+                }
+
+        return None
+
+
+
+def to_anthropic_format(tools: List[ToolDef]) -> List[Dict[str, Any]]:
+    '''Convert ToolDef objects to Anthropic tool format.
+    
+    Args:
+        tools: List of ToolDef objects to convert
+        
+    Returns:
+        List of dictionaries in Anthropic tool format
+    '''
+
+    anthropic_tools = []
+    for tool in tools:
+        anthropic_tool = {
+            'name': tool.name,
+            'description': tool.description,
+            'input_schema': {
+                'type': 'object',
+                'properties': {},
+                'required': []
+            }
+        }
+
+        # Add properties
+        for param in tool.parameters:
+            # Map the type or use the original if no mapping exists
+            schema_type = TYPE_MAPPING.get(param.parameter_type, param.parameter_type)
+
+            param_schema = {
+                'type': schema_type,  # Use mapped type
+                'description': param.description
+            }
+
+            # For arrays, we need to specify the items type
+            if schema_type == 'array':
+                item_type = _infer_array_item_type(param)
+                param_schema['items'] = {'type': item_type}
+
+            anthropic_tool['input_schema']['properties'][param.name] = param_schema
+
+            # Add default value if provided
+            if param.default is not None:
+                anthropic_tool['input_schema']['properties'][param.name]['default'] = param.default
+
+            # Add to required list if required
+            if param.required:
+                anthropic_tool['input_schema']['required'].append(param.name)
+
+        anthropic_tools.append(anthropic_tool)
+    return anthropic_tools
+
+
+def _infer_array_item_type(param: ToolParameter) -> str:
+    '''Infer the item type for an array parameter based on its name and description.
+    
+    Args:
+        param: The ToolParameter object
+        
+    Returns:
+        The inferred JSON Schema type for array items
+    '''
+    # Default to string items
+    item_type = 'string'
+
+    # Check if parameter name contains hints about item type
+    param_name_lower = param.name.lower()
+    if any(hint in param_name_lower for hint in ['language', 'code', 'tag', 'name', 'id']):
+        item_type = 'string'
+    elif any(hint in param_name_lower for hint in ['number', 'count', 'amount', 'index']):
+        item_type = 'integer'
+
+    # Also check the description for hints
+    if param.description:
+        desc_lower = param.description.lower()
+        if 'string' in desc_lower or 'text' in desc_lower or 'language' in desc_lower:
+            item_type = 'string'
+        elif 'number' in desc_lower or 'integer' in desc_lower or 'int' in desc_lower:
+            item_type = 'integer'
+
+    return item_type

classes/client.py → client/mcp_client.py

@@ -46,6 +46,18 @@ class ToolDef:
     identifier: str = ''
 
 
+@dataclass
+class ToolInvocationResult:
+    '''Represents the result of a tool invocation.
+    
+    Attributes:
+        content: Result content as a string
+        error_code: Error code (0 for success, 1 for error)
+    '''
+    content: str
+    error_code: int
+
+
 class MCPConnectionError(Exception):
     '''Exception raised when MCP connection fails'''
     pass
@@ -68,11 +80,11 @@ class MCPClientWrapper:
             max_retries: Maximum number of retry attempts
         '''
 
-        if urlparse(endpoint).scheme not in ('http', 'https'):
-            raise ValueError(f'Endpoint {endpoint} is not a valid HTTP(S) URL')
         self.endpoint = endpoint
         self.timeout = timeout
         self.max_retries = max_retries
+        # self.tools = None
+        # self.anthropic = Anthropic()
 
 
     async def _execute_with_retry(self, operation_name: str, operation_func):
@@ -136,6 +148,7 @@ class MCPClientWrapper:
                 f'{operation_name} failed after {self.max_retries} attempts: {str(last_exception)}'
             )
 
+
     async def _safe_sse_operation(self, operation_func):
         '''Safely execute an SSE operation with proper task cleanup
         
@@ -184,6 +197,7 @@ class MCPClientWrapper:
                     logger.warning('Error during task cleanup: %s', cleanup_error)
             raise
 
+
     async def list_tools(self) -> List[ToolDef]:
         '''List available tools from the MCP endpoint
         
@@ -194,6 +208,7 @@ class MCPClientWrapper:
             MCPConnectionError: If connection fails
             MCPTimeoutError: If operation times out
         '''
+
         async def _list_tools_operation():
             async def _operation(session):
 
@@ -224,8 +239,74 @@ class MCPClientWrapper:
                             identifier=tool.name  # Using name as identifier
                         )
                     )
+
+                self.tools = tools
+
                 return tools
 
             return await self._safe_sse_operation(_operation)
 
         return await self._execute_with_retry('list_tools', _list_tools_operation)
+
+
+    async def invoke_tool(self, tool_name: str, kwargs: Dict[str, Any]) -> ToolInvocationResult:
+        '''Invoke a specific tool with parameters
+        
+        Args:
+            tool_name: Name of the tool to invoke
+            kwargs: Dictionary of parameters to pass to the tool
+            
+        Returns:
+            ToolInvocationResult containing the tool's response
+            
+        Raises:
+            MCPConnectionError: If connection fails
+            MCPTimeoutError: If operation times out
+        '''
+
+        async def _invoke_tool_operation():
+            async def _operation(session):
+                result = await session.call_tool(tool_name, kwargs)
+                return ToolInvocationResult(
+                    content='\n'.join([result.model_dump_json() for result in result.content]),
+                    error_code=1 if result.isError else 0,
+                )
+
+            return await self._safe_sse_operation(_operation)
+
+        return await self._execute_with_retry(f'invoke_tool({tool_name})', _invoke_tool_operation)
+
+
+    async def check_connection(self) -> bool:
+        '''Check if the MCP endpoint is reachable
+        
+        Returns:
+            True if connection is successful, False otherwise
+        '''
+
+        logger = logging.getLogger(__name__ + '_check_connection')
+
+        try:
+            await self.list_tools()
+            return True
+        except Exception as e: # pylint: disable=broad-exception-caught
+            logger.debug('Connection check failed: %s', str(e))
+            return False
+
+
+    def get_endpoint_info(self) -> Dict[str, Any]:
+        '''Get information about the configured endpoint
+        
+        Returns:
+            Dictionary with endpoint information
+        '''
+        parsed = urlparse(self.endpoint)
+        return {
+            'endpoint': self.endpoint,
+            'scheme': parsed.scheme,
+            'hostname': parsed.hostname,
+            'port': parsed.port,
+            'path': parsed.path,
+            'timeout': self.timeout,
+            'max_retries': self.max_retries
+        }

rss_client.py

@@ -1,11 +1,15 @@
 '''RSS MCP server demonstration client app.'''
 
+import os
+import json
 import logging
 from pathlib import Path
 from logging.handlers import RotatingFileHandler
 
 import gradio as gr
-from classes.client import MCPClientWrapper
+from gradio.components.chatbot import ChatMessage
+from client.mcp_client import MCPClientWrapper
+from client.anthropic_bridge import AnthropicBridge
 
 # Make sure log directory exists
 Path('logs').mkdir(parents=True, exist_ok=True)
@@ -15,7 +19,7 @@ logger = logging.getLogger()
 
 logging.basicConfig(
     handlers=[RotatingFileHandler(
-        'logs/rss_server.log',
+        'logs/rss_client.log',
         maxBytes=100000,
         backupCount=10,
         mode='w'
@@ -27,17 +31,99 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 client = MCPClientWrapper('https://agents-mcp-hackathon-rss-mcp-server.hf.space/gradio_api/mcp/sse')
+bridge = AnthropicBridge(
+    client,
+    api_key=os.environ['ANTHROPIC_API_KEY']
+)
+
+async def submit_input(message: str, chat_history: list) -> str:
+    '''Submits user message to agent'''
+
+    function_logger = logging.getLogger(__name__ + '.submit_input')
+
+    chat_history.append({"role": "user", "content": message})
+    input_messages = format_chat_history(chat_history)
+    function_logger.info(input_messages)
+
+    result = await bridge.process_query(input_messages)
+    function_logger.info(result)
+    function_logger.info(result.keys())
+
+    try:
+        chat_history.append({
+            "role": "assistant",
+            "content": result['llm_response'].content[0].text
+        })
+
+    except AttributeError:
+        function_logger.info('Model called the tool, but did not talk about it')
+
+    if result['tool_result']:
+        articles = json.loads(result['tool_result'].content)['text']
+        function_logger.info(articles)
+        tmp_chat_history = chat_history.copy()
+        tmp_chat_history.append({
+            "role": "assistant",
+            "content": ('Here are the three most recent entries from the RSS ' +
+                f'feed in JSON format. Tell the user what you have found: {json.dumps(articles)}')
+        })
+
+        tmp_input_messages = format_chat_history(tmp_chat_history)
+        function_logger.info(tmp_input_messages)
+        result = await bridge.process_query(tmp_input_messages)
+
+        chat_history.append({
+            "role": "assistant",
+            "content": result['llm_response'].content[0].text
+        })
+
+
+    return '', chat_history
+
+
+def format_chat_history(history) -> list[dict]:
+    '''Formats gradio chat history for submission to anthropic.'''
+
+    messages = []
+
+    for chat_message in history:
+        if isinstance(msg, ChatMessage):
+            role, content = chat_message.role, chat_message.content
+        else:
+            role, content = chat_message.get("role"), chat_message.get("content")
+
+        if role in ["user", "assistant", "system"]:
+            messages.append({"role": role, "content": content})
+
+    return messages
+
 
 with gr.Blocks(title='MCP RSS client') as demo:
     gr.Markdown('# MCP RSS reader')
     gr.Markdown(
-        'Connect to the MCP RSS server: https://huggingface.co/spaces/Agents-MCP-Hackathon/rss-mcp-server'
+        'Connect to the MCP RSS server: ' +
+        'https://huggingface.co/spaces/Agents-MCP-Hackathon/rss-mcp-server'
     )
 
     connect_btn = gr.Button('Connect')
-    status = gr.Textbox(label='Connection Status', interactive=False, lines=50)
-    connect_btn.click(client.list_tools, outputs=status) # pylint: disable=no-member
+    status = gr.Textbox(label='Connection Status', interactive=False, lines=10)
+
+    chatbot = gr.Chatbot(
+        value=[],
+        height=500,
+        type='messages',
+        show_copy_button=True,
+        avatar_images=('👤', '🤖')
+    )
 
+    msg = gr.Textbox(
+        label='Your Question',
+        placeholder='Ask about an RSS feed',
+        scale=4
+    )
+
+    connect_btn.click(client.list_tools, outputs=status) # pylint: disable=no-member
+    msg.submit(submit_input, [msg, chatbot], [msg, chatbot]) # pylint: disable=no-member
 
 if __name__ == '__main__':