Lenses MCP Server

Instructions

Implementation Reference

• src/lenses_mcp/tools/kafka_connectors.py:102-127 (handler)

  The core handler function decorated with @mcp.tool(), implementing the logic to set an action (start, stop, restart, pause, resume) on a Kafka connector by making a PUT request to the Lenses API endpoint.

  @mcp.tool()
  async def set_action_on_kafka_connector(
      environment: str,
      cluster: str,
      connector: str,
      action: str
  ) -> Dict[str, Any]:
      """
      Controls a Kafka connector (start, stop, restart, pause, resume).
      
      Args:
          environment: The environment name.
          cluster: The cluster name.
          connector: The connector name.
          action: The action to perform. Options: "start", "stop", "restart", "pause", "resume".
      
      Returns:
          The result of the control operation.
      """
      valid_actions = ["start", "stop", "restart", "pause", "resume"]
      if action not in valid_actions:
          raise ValueError(f"Action must be one of: {', '.join(valid_actions)}")
      
      endpoint = f"/api/v1/environments/{environment}/proxy/api/kafka-connect/clusters/{cluster}/connectors/{connector}/{action}"
      return await api_client._make_request("PUT", endpoint)

• src/lenses_mcp/server.py:28-33 (registration)

  The registration block in the main MCP server file where register_kafka_connectors(mcp) is called, which in turn defines and registers the set_action_on_kafka_connector tool using @mcp.tool() decorator.

  register_environments(mcp)
  register_kafka_connectors(mcp)
  register_kafka_consumer_groups(mcp)
  register_sql(mcp)
  register_sql_processors(mcp)
  register_topics(mcp)