MCP with Gemini Integration

import logging import inspect from typing import Any from mcp.server.fastmcp import FastMCP def make_tool_wrapper(tool_def): """ Dynamically create a wrapper function with the correct signature for MCP registration. """ params = tool_def.input_schema.get('properties', {}) required = set(tool_def.input_schema.get('required', [])) # Build the function signature parameters = [] for name, prop in params.items(): annotation = str if prop.get('type') == 'integer': annotation = int elif prop.get('type') == 'number': annotation = float param = inspect.Parameter( name, inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=annotation, default=inspect.Parameter.empty if name in required else None ) parameters.append(param) async def wrapper_func(*args, **kwargs): # Map args to parameter names bound_args = {} for i, name in enumerate(params.keys()): if i < len(args): bound_args[name] = args[i] elif name in kwargs: bound_args[name] = kwargs[name] else: bound_args[name] = None # Call the tool handler with correct arguments return await tool_def.handler(**bound_args) # Set the correct signature wrapper_func.__signature__ = inspect.Signature(parameters) wrapper_func.__name__ = tool_def.name wrapper_func.__qualname__ = tool_def.name wrapper_func.__doc__ = tool_def.description return wrapper_func def register_dynamic_tools(mcp: FastMCP, logger: logging.Logger) -> int: """Register all available tools dynamically with correct signatures.""" from src.tools import ToolRegistry try: registry = ToolRegistry() registered_tools = registry.discover_and_register_tools() if not registered_tools: raise RuntimeError("No tools were discovered. Please check that tools exist in src/tools/ directory and have register_tool() functions.") logger.info(f"Successfully discovered {len(registered_tools)} tools: {[tool.name for tool in registered_tools]}") registered_count = 0 for tool_def in registered_tools: try: wrapper = make_tool_wrapper(tool_def) mcp.tool()(wrapper) logger.info(f"Registered tool: {tool_def.name} - {tool_def.description}") registered_count += 1 except Exception as e: logger.error(f"Failed to register tool '{tool_def.name}': {e}") continue if registered_count == 0: raise RuntimeError("No tools were successfully registered") logger.info(f"Successfully registered {registered_count} out of {len(registered_tools)} discovered tools") return registered_count except Exception as e: logger.error(f"Error during tool registration: {e}") raise RuntimeError(f"Tool registration failed: {str(e)}")